Merge ab92678 into d57a2c8

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@luckbox/mq-clients",
-  "version": "1.0.8",
+  "version": "1.0.9",
   "main": "dist/index.js",
   "types": "dist/index.d.ts",
   "scripts": {

src/rabbitmq/RabbitMQClient.ts

@@ -87,7 +87,7 @@ class RabbitMQClient implements MQClient {
   async subscribe(namespace: string, callback: CallbackFunc) {
     assert(this.connection, 'You must connect() first!');
 
-    await this.saveSubscription(namespace, callback);
+    await this.saveSubscription(this.isExchangeInDirectType() ? this.exchangeConfig.name : namespace, callback);
     await this.createExchange(namespace);
     await this.createQueueAndBindItToExchange(namespace);
   }
@@ -158,14 +158,12 @@ class RabbitMQClient implements MQClient {
   }
 
   private saveSubscription(namespace: string, callback: CallbackFunc) {
-    if (!this.subscriptions.has(namespace)) {
-      this.subscriptions.set(namespace, []);
-    }
+    const callbacks = this.subscriptions.has(namespace)
+      ? this.subscriptions.get(namespace)
+      : [];
+    callbacks.push(callback);
 
-    this.subscriptions.set(namespace, [
-      ...this.subscriptions.get(namespace),
-      callback,
-    ]);
+    this.subscriptions.set(namespace, callbacks);
   }
 
   private async createExchange(exchange: string) {

tests/RabbitMQClient.test.ts

@@ -228,16 +228,22 @@ describe('RabbitMQClient', () => {
     });
 
     it('should invoke the callback passed to subscribe when a message is received', async () => {
+      const directModeConfiguredClient = new RabbitMQClient(createConstructorParams({
+        exchange: {
+          type: 'direct',
+          name: 'my-exchange',
+        },
+      }));
       const mockQueue = createMockQueue();
       mockChannel.assertQueue.mockResolvedValueOnce(mockQueue);
 
-      await client.connect();
+      await directModeConfiguredClient.connect();
 
       const mockCallback = jest.fn();
       const dummyData = { bar: 'foo' };
-      await client.subscribe('my-namespace', mockCallback);
+      await directModeConfiguredClient.subscribe('my-namespace', mockCallback);
 
-      triggerMockChannelConsumer('my-namespace', mockQueue.queue, dummyData);
+      triggerMockChannelConsumer('my-exchange', mockQueue.queue, dummyData);
       await sleep(testRetryTimeout);
 
       expect(mockCallback).toHaveBeenCalledWith(dummyData);
@@ -317,7 +323,7 @@ describe('RabbitMQClient', () => {
       expect(mockCallback).toHaveBeenCalledWith(dummyData);
     });
 
-    it('should not invoke the callback passed to subscribe when a maformed message is received after a reconnect', async () => {
+    it('should not invoke the callback passed to subscribe when a malformed message is received after a reconnect', async () => {
       await client.connect();
 
       const mockCallback = jest.fn();