feat(rabbitmq): add exchange-to-exchange bindings config

integration/rabbitmq/e2e/configuration.e2e-spec.ts

@@ -582,5 +582,85 @@ describe('Module Configuration', () => {
         await app.close();
       });
     });
+
+    it('should create exchange bindings', async () => {
+      const originalConnect = amqplib.connect;
+      let bindExchangeSpy;
+
+      const connectSpy = jest
+        .spyOn(amqplib, 'connect')
+        .mockImplementation((...args) => {
+          const result = originalConnect(...args);
+          result.then((conn) => {
+            const originalCreateConfirmChannel = conn.createConfirmChannel;
+            jest
+              .spyOn(conn, 'createConfirmChannel')
+              .mockImplementation(function (...args) {
+                const result = originalCreateConfirmChannel.apply(this, args);
+                result.then((channel) => {
+                  bindExchangeSpy = jest.spyOn(channel, 'bindExchange');
+                });
+                return result;
+              });
+          });
+          return result;
+        });
+
+      const otherExchangeName = 'otherExchange';
+
+      app = await Test.createTestingModule({
+        imports: [
+          RabbitMQModule.forRootAsync(RabbitMQModule, {
+            useFactory: async () => {
+              return {
+                exchanges: [
+                  {
+                    name: nonExistingExchange,
+                    type: 'topic',
+                    createExchangeIfNotExists: true,
+                  },
+                  {
+                    name: otherExchangeName,
+                    type: 'topic',
+                    createExchangeIfNotExists: true,
+                  },
+                ],
+                exchangeBindings: [
+                  {
+                    destination: otherExchangeName,
+                    source: nonExistingExchange,
+                    pattern: '*',
+                  },
+                ],
+                uri,
+                connectionInitOptions: {
+                  wait: true,
+                  reject: true,
+                  timeout: 3000,
+                },
+              };
+            },
+          }),
+        ],
+      }).compile();
+
+      const amqpConnection = app.get<AmqpConnection>(AmqpConnection);
+      expect(app).toBeDefined();
+
+      expect(connectSpy).toHaveBeenCalledTimes(1);
+      expect(connectSpy).toHaveBeenCalledWith(amqplibUri, undefined);
+      expect(bindExchangeSpy).toHaveBeenCalledWith(
+        otherExchangeName,
+        nonExistingExchange,
+        '*',
+        undefined,
+      );
+
+      await app.init();
+      expect(
+        await amqpConnection.channel.checkExchange(nonExistingExchange),
+      ).toBeDefined();
+      await app.close();
+    });
   });
 });

packages/rabbitmq/src/amqp/connection.ts

@@ -90,6 +90,7 @@ const defaultConfig = {
   ),
   defaultSubscribeErrorBehavior: MessageHandlerErrorBehavior.REQUEUE,
   exchanges: [],
+  exchangeBindings: [],
   queues: [],
   defaultRpcTimeout: 10000,
   connectionInitOptions: {
@@ -291,6 +292,17 @@ export class AmqpConnection {
         })
       );
 
+      await Promise.all(
+        this.config.exchangeBindings.map((exchangeBinding) =>
+          channel.bindExchange(
+            exchangeBinding.destination,
+            exchangeBinding.source,
+            exchangeBinding.pattern,
+            exchangeBinding.args
+          )
+        )
+      );
+
       await this.setupQueuesWithBindings(channel, this.config.queues);
 
       if (this.config.enableDirectReplyTo) {

packages/rabbitmq/src/rabbitmq.interfaces.ts

@@ -23,6 +23,13 @@ export interface RabbitMQQueueConfig {
   bindQueueArguments?: any;
 }
 
+export interface RabbitMQExchangeBindingConfig {
+  destination: string;
+  source: string;
+  pattern: string;
+  args?: any;
+}
+
 export type ConsumeOptions = Options.Consume;
 
 export interface MessageOptions {
@@ -110,6 +117,7 @@ export interface RabbitMQConfig {
    */
   prefetchCount?: number;
   exchanges?: RabbitMQExchangeConfig[];
+  exchangeBindings?: RabbitMQExchangeBindingConfig[];
   queues?: RabbitMQQueueConfig[];
   defaultRpcTimeout?: number;
   defaultExchangeType?: string;