Add dynamic queues to setup function in ChannelWrapper

.github/workflows/publish.yaml

@@ -2,8 +2,9 @@ name: Publish NPM package
 
 on:
   push:
-    branches:
-      - main
+    branches: [main]
+  pull_request:
+    branches: [main]
 
 env:
   CLOUDAMQP_API_KEY: ${{ secrets.CLOUDAMQP_API_KEY }}
@@ -21,5 +22,6 @@ jobs:
       - run: npm run build
       - run: npm test
       - uses: JS-DevTools/npm-publish@v1
+        if: ${{ github.ref == 'refs/heads/main' }}
         with:
           token: ${{ env.NPM_TOKEN }}

src/channel.ts

@@ -1,6 +1,5 @@
 import { MessageBusConfig } from 'Types';
 import { error, log } from 'Utils';
-import { ConfirmChannel } from 'amqplib';
 import { configureMessageBus, configureMessageBusStatic } from './config';
 import {
   closeMessageBusConnection,
@@ -34,29 +33,30 @@ const channelPromise = getConnection()
   .then((connection) => {
     return connection.createChannel({
       json: true,
-      setup: async (channel: ConfirmChannel) => {
-        log(
-          `Waiting for AMQP to be initialized by invoking initMessageBus() from 'node-message-bus'.`
-        );
-        const config = await _initPromise;
-
-        await configureMessageBus(config, channel);
-
-        log(
-          `AMQP initialization is complete with the following config: ${JSON.stringify(
-            config
-          )}`
-        );
-
-        postInitPromiseResolve(true);
-      },
     });
   })
   .catch((e) => {
     error(e.stack || e);
     throw e;
   });
 
+channelPromise.then(async (channelWrapper) => {
+  log(
+    `Waiting for AMQP to be initialized by invoking initMessageBus() from 'node-message-bus'.`
+  );
+  const config = await _initPromise;
+
+  await configureMessageBus(config, channelWrapper);
+
+  log(
+    `AMQP initialization is complete with the following config: ${JSON.stringify(
+      config
+    )}`
+  );
+
+  postInitPromiseResolve(true);
+});
+
 export const getDefaultChannel = async () => {
   const channel = await channelPromise;
   await channel.waitForConnect();

src/config.ts

@@ -69,7 +69,7 @@ export const configureMessageBusStatic = async (
  */
 export const configureMessageBus = async (
   config: MessageBusConfig,
-  channel?: ChannelWrapper | ConfirmChannel
+  channel?: ChannelWrapper
 ) => {
   configureMessageBusStatic(config);
 
@@ -85,9 +85,17 @@ export const configureMessageBus = async (
   )) {
     defaultExchangeConfigured = true;
     try {
-      log(`Asserting exchange "${exchange.name}" of type "${exchange.type}".`);
       promises.push(
-        channel.assertExchange(exchange.name, exchange.type, exchange.options)
+        channel.addSetup(async (channel: ConfirmChannel) => {
+          log(
+            `Asserting exchange "${exchange.name}" of type "${exchange.type}".`
+          );
+          await channel.assertExchange(
+            exchange.name,
+            exchange.type,
+            exchange.options
+          );
+        })
       );
       appliedConfig.exchanges.push(exchange);
     } catch (e) {
@@ -102,8 +110,12 @@ export const configureMessageBus = async (
     DEFAULT_CONFIG.queues || []
   )) {
     try {
-      log(`Asserting queue "${queue.name}".`);
-      promises.push(channel.assertQueue(queue.name, queue.options));
+      promises.push(
+        channel.addSetup(async (channel: ConfirmChannel) => {
+          log(`Asserting queue "${queue.name}".`);
+          await channel.assertQueue(queue.name, queue.options);
+        })
+      );
       appliedConfig.queues.push(queue);
     } catch (e) {
       const message = `Failed to assert queue "${
@@ -118,11 +130,17 @@ export const configureMessageBus = async (
   )) {
     const fromExchange = binding.fromExchange || DEFAULT_EXCHANGE_NAME;
     try {
-      log(
-        `Asserting the queue binding "${fromExchange}" -> "${binding.toQueue}" by key "${binding.routingKey}".`
-      );
       promises.push(
-        channel.bindQueue(binding.toQueue, fromExchange, binding.routingKey)
+        channel.addSetup(async (channel: ConfirmChannel) => {
+          log(
+            `Asserting the queue binding "${fromExchange}" -> "${binding.toQueue}" by key "${binding.routingKey}".`
+          );
+          await channel.bindQueue(
+            binding.toQueue,
+            fromExchange,
+            binding.routingKey
+          );
+        })
       );
       appliedConfig.bindings.push(binding);
     } catch (e) {

src/consumer.ts

@@ -118,7 +118,9 @@ export async function consumeMessages<Message extends IMessage>(
   queueName: ConsumeMessagesConfig | string,
   handler: MessageHandler<Message>
 ) {
+  console.log('📍 consumeMessages', queueName);
   let prefetch: number | undefined;
+  const channel = await getDefaultChannel();
   if (typeof queueName !== 'string') {
     if (!queueName.queues || queueName.queues.length !== 1) {
       throw new Error(
@@ -133,7 +135,6 @@ export async function consumeMessages<Message extends IMessage>(
     }
   }
 
-  const channel = await getDefaultChannel();
   const queue = getMessageBusConfig().queues.find((q) => q.name === queueName);
   const consumerTag = `${
     process.env.HOSTNAME || 'no.env.HOSTNAME'