diff --git a/.github/workflows/publish.yaml b/.github/workflows/publish.yaml index 5f163c1..ddccc7e 100644 --- a/.github/workflows/publish.yaml +++ b/.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 }} diff --git a/package-lock.json b/package-lock.json index 7f4037d..53ce98c 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "node-message-bus", - "version": "3.3.3", + "version": "3.3.4", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "node-message-bus", - "version": "3.3.3", + "version": "3.3.4", "license": "MIT", "dependencies": { "@types/amqplib": "^0.10.5", diff --git a/package.json b/package.json index a71d838..37cbc9a 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "node-message-bus", - "version": "3.3.4", + "version": "3.3.5", "description": "Minimalistic and complete AMQP message bus implementation", "main": "lib/index.js", "files": [ diff --git a/src/channel.ts b/src/channel.ts index 1edb2bb..cfa7da2 100644 --- a/src/channel.ts +++ b/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,22 +33,6 @@ 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) => { @@ -57,6 +40,23 @@ const channelPromise = getConnection() 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(); diff --git a/src/config.ts b/src/config.ts index 271e670..e64cd18 100644 --- a/src/config.ts +++ b/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) {