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();
+    });
   });
 });

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

@@ -15,6 +15,8 @@ const exchange = 'testSubscribeExchange';
 const routingKey1 = 'testSubscribeRoute1';
 const routingKey2 = 'testSubscribeRoute2';
 const routingKey3 = 'testSubscribeViaHandlerRoute';
+const routingKey4 = 'testSubscribeViaHandlerRouteMulti1';
+const routingKey5 = 'testSubscribeViaHandlerRouteMulti2';
 const nonJsonRoutingKey = 'nonJsonSubscribeRoute';
 
 const createRoutingKey = 'test.create.object';
@@ -40,6 +42,13 @@ class SubscribeService {
     testHandler(message);
   }
 
+  @RabbitSubscribe({
+    name: 'handler2',
+  })
+  handleSubscribeByNameMulti(message: object) {
+    testHandler(message);
+  }
+
   @RabbitSubscribe({
     exchange,
     routingKey: [routingKey1, routingKey2],
@@ -152,6 +161,16 @@ describe('Rabbit Subscribe', () => {
               exchange,
               routingKey: [routingKey3],
             },
+            handler2: [
+              {
+                exchange,
+                routingKey: routingKey4,
+              },
+              {
+                exchange,
+                routingKey: routingKey5,
+              },
+            ],
           },
           uri,
           connectionInitOptions: { wait: true, reject: true, timeout: 3000 },
@@ -196,6 +215,17 @@ describe('Rabbit Subscribe', () => {
     expect(testHandler).toHaveBeenCalledWith(`testMessage`);
   });
 
+  it('should receive all messages when subscribed via handler name with multiple configs', async () => {
+    await amqpConnection.publish(exchange, routingKey4, 'testMessage');
+    await amqpConnection.publish(exchange, routingKey5, 'testMessage2');
+
+    await new Promise((resolve) => setTimeout(resolve, 50));
+
+    expect(testHandler).toHaveBeenCalledTimes(2);
+    expect(testHandler).toHaveBeenCalledWith(`testMessage`);
+    expect(testHandler).toHaveBeenCalledWith(`testMessage2`);
+  });
+
   it('should work with a topic exchange set up that has multiple subscribers with similar routing keys', async () => {
     const routingKeys = [createRoutingKey, updateRoutingKey, deleteRoutingKey];
 

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 {
@@ -97,7 +104,10 @@ export interface ConnectionInitOptions {
 }
 
 export type RabbitMQChannels = Record<string, RabbitMQChannelConfig>;
-export type RabbitMQHandlers = Record<string, MessageHandlerOptions>;
+export type RabbitMQHandlers = Record<
+  string,
+  MessageHandlerOptions | MessageHandlerOptions[]
+>;
 
 export interface RabbitMQConfig {
   name?: string;
@@ -107,6 +117,7 @@ export interface RabbitMQConfig {
    */
   prefetchCount?: number;
   exchanges?: RabbitMQExchangeConfig[];
+  exchangeBindings?: RabbitMQExchangeBindingConfig[];
   queues?: RabbitMQQueueConfig[];
   defaultRpcTimeout?: number;
   defaultExchangeType?: string;

packages/rabbitmq/src/rabbitmq.module.ts

@@ -1,4 +1,9 @@
-import { DiscoveryModule, DiscoveryService } from '@golevelup/nestjs-discovery';
+import {
+  DiscoveredMethod,
+  DiscoveredMethodWithMeta,
+  DiscoveryModule,
+  DiscoveryService,
+} from '@golevelup/nestjs-discovery';
 import {
   createConfigurableDynamicRootModule,
   IConfigurableDynamicRootModule,
@@ -135,6 +140,40 @@ export class RabbitMQModule
     RabbitMQModule.bootstrapped = false;
   }
 
+  private async setupHandler(
+    connection: AmqpConnection,
+    discoveredMethod: DiscoveredMethod,
+    config: RabbitHandlerConfig,
+    handler: (...args: any[]) => Promise<any>
+  ) {
+    const handlerDisplayName = `${discoveredMethod.parentClass.name}.${
+      discoveredMethod.methodName
+    } {${config.type}} -> ${
+      // eslint-disable-next-line sonarjs/no-nested-template-literals
+      config.queueOptions?.channel ? `${config.queueOptions.channel}::` : ''
+    }${config.exchange}::${config.routingKey}::${config.queue || 'amqpgen'}`;
+
+    if (
+      config.type === 'rpc' &&
+      !connection.configuration.enableDirectReplyTo
+    ) {
+      this.logger.warn(
+        `Direct Reply-To Functionality is disabled. RPC handler ${handlerDisplayName} will not be registered`
+      );
+      return;
+    }
+
+    this.logger.log(handlerDisplayName);
+
+    return config.type === 'rpc'
+      ? connection.createRpc(handler, config)
+      : connection.createSubscriber(
+          handler,
+          config,
+          discoveredMethod.methodName
+        );
+  }
+
   // eslint-disable-next-line sonarjs/cognitive-complexity
   public async onApplicationBootstrap() {
     if (RabbitMQModule.bootstrapped) {
@@ -199,38 +238,28 @@ export class RabbitMQModule
               'rmq' // contextType
             );
 
-            const mergedConfig = {
-              ...config,
-              ...connection.configuration.handlers[config.name || ''],
-            };
-            const { exchange, routingKey, queue, queueOptions } = mergedConfig;
+            const moduleHandlerConfigRaw =
+              connection.configuration.handlers[config.name || ''];
 
-            const handlerDisplayName = `${discoveredMethod.parentClass.name}.${
-              discoveredMethod.methodName
-            } {${config.type}} -> ${
-              // eslint-disable-next-line sonarjs/no-nested-template-literals
-              queueOptions?.channel ? `${queueOptions.channel}::` : ''
-            }${exchange}::${routingKey}::${queue || 'amqpgen'}`;
-
-            if (
-              config.type === 'rpc' &&
-              !connection.configuration.enableDirectReplyTo
-            ) {
-              this.logger.warn(
-                `Direct Reply-To Functionality is disabled. RPC handler ${handlerDisplayName} will not be registered`
-              );
-              return;
-            }
+            const moduleHandlerConfigs = Array.isArray(moduleHandlerConfigRaw)
+              ? moduleHandlerConfigRaw
+              : [moduleHandlerConfigRaw];
 
-            this.logger.log(handlerDisplayName);
+            await Promise.all(
+              moduleHandlerConfigs.map((moduleHandlerConfig) => {
+                const mergedConfig = {
+                  ...config,
+                  ...moduleHandlerConfig,
+                };
 
-            return config.type === 'rpc'
-              ? connection.createRpc(handler, mergedConfig)
-              : connection.createSubscriber(
-                  handler,
+                return this.setupHandler(
+                  connection,
+                  discoveredMethod,
                   mergedConfig,
-                  discoveredMethod.methodName
+                  handler
                 );
+              })
+            );
           })
         );
       }