Add dynamic queues to setup function in ChannelWrapper

nikitaeverywhere · Jun 11, 2024 · 3e21e96 · 3e21e96
1 parent d98e132
commit 3e21e96
Show file tree

Hide file tree

Showing 2 changed files with 7 additions and 4 deletions.
diff --git a/package-lock.json b/package-lock.json
diff --git a/src/consumer.ts b/src/consumer.ts
@@ -13,7 +13,7 @@ import {
   pushToLastRejectedMessages,
 } from 'Utils';
 import { ChannelWrapper } from 'amqp-connection-manager';
-import { ConsumeMessage } from 'amqplib';
+import { ConfirmChannel, ConsumeMessage } from 'amqplib';
 import { DEFAULT_EXCHANGE_NAME } from './Const';
 import { getDefaultChannel } from './channel';
 import { configureMessageBus, getMessageBusConfig } from './config';
@@ -119,6 +119,7 @@ export async function consumeMessages<Message extends IMessage>(
   handler: MessageHandler<Message>
 ) {
   let prefetch: number | undefined;
+  const channel = await getDefaultChannel();
   if (typeof queueName !== 'string') {
     if (!queueName.queues || queueName.queues.length !== 1) {
       throw new Error(
@@ -131,9 +132,11 @@ export async function consumeMessages<Message extends IMessage>(
     if (prefetchCount) {
       prefetch = prefetchCount;
     }
+    channel.addSetup(async (channel: ConfirmChannel) => {
+      await configureMessageBus(config, channel);
+    });
   }
 
-  const channel = await getDefaultChannel();
   const queue = getMessageBusConfig().queues.find((q) => q.name === queueName);
   const consumerTag = `${
     process.env.HOSTNAME || 'no.env.HOSTNAME'