feat(rabbitmq): bind handlerless queues to exchange

golevelup · Nov 7, 2023 · e711bdc · e711bdc
1 parent 086bf69
commit e711bdc
Show file tree

Hide file tree

Showing 2 changed files with 43 additions and 9 deletions.
diff --git a/packages/rabbitmq/src/amqp/connection.ts b/packages/rabbitmq/src/amqp/connection.ts
@@ -22,7 +22,7 @@ import {
 } from 'rxjs';
 import { catchError, filter, first, map, take, timeout } from 'rxjs/operators';
 import { randomUUID } from 'crypto';
-import { defaultAssertQueueErrorHandler } from '..';
+import { defaultAssertQueueErrorHandler, RabbitMQQueueConfig } from '..';
 import {
   ConnectionInitOptions,
   MessageHandlerOptions,
@@ -281,14 +281,7 @@ export class AmqpConnection {
         return channel.checkExchange(x.name);
       });
 
-      this.config.queues.forEach((queue) => {
-        const { createQueueIfNotExists = true } = queue;
-
-        if (createQueueIfNotExists) {
-          return channel.assertQueue(queue.name, queue.options);
-        }
-        return channel.checkQueue(queue.name);
-      });
+      this.setupQueuesWithBindings(channel, this.config.queues);
 
       if (this.config.enableDirectReplyTo) {
         await this.initDirectReplyQueue(channel);
@@ -298,6 +291,44 @@ export class AmqpConnection {
     }
   }
 
+  private async setupQueuesWithBindings(
+    channel: ConfirmChannel,
+    queues: RabbitMQQueueConfig[]
+  ) {
+    for (const configuredQueue of queues) {
+      const { createQueueIfNotExists = true } = configuredQueue;
+
+      if (createQueueIfNotExists) {
+        this.setupQueue(configuredQueue, channel);
+      }
+      await channel.assertQueue(configuredQueue.name, configuredQueue.options);
+
+      let routingKeys: string[] = [];
+      if (Array.isArray(configuredQueue.routingKey)) {
+        routingKeys = configuredQueue.routingKey;
+      } else {
+        if (configuredQueue.routingKey != null) {
+          routingKeys = [configuredQueue.routingKey];
+        }
+      }
+
+      if (routingKeys.length > 0) {
+        await Promise.all(
+          routingKeys.map((routingKey) => {
+            if (configuredQueue.exchange != undefined) {
+              channel.bindQueue(
+                configuredQueue.name,
+                configuredQueue.exchange,
+                routingKey,
+                configuredQueue.bindQueueArguments
+              );
+            }
+          })
+        );
+      }
+    }
+  }
+
   private async initDirectReplyQueue(channel: ConfirmChannel) {
     // Set up a consumer on the Direct Reply-To queue to facilitate RPC functionality
     await channel.consume(

diff --git a/packages/rabbitmq/src/rabbitmq.interfaces.ts b/packages/rabbitmq/src/rabbitmq.interfaces.ts
@@ -18,6 +18,9 @@ export interface RabbitMQQueueConfig {
   name: string;
   createQueueIfNotExists?: boolean;
   options?: Options.AssertQueue;
+  exchange?: string;
+  routingKey?: string | string[];
+  bindQueueArguments?: any;
 }
 
 export type ConsumeOptions = Options.Consume;