bugfix(microservices) fix rabbit mq transport, code refactor, add tests

Author: kamilmysliwiec

integration/docker-compose.yml

@@ -36,4 +36,11 @@ services:
     environment:
       - MONGODB_DATABASE="test"
     ports:
-      - 27017:27017
+      - 27017:27017
+  rabbit:
+    hostname: rabbit
+    image: "rabbitmq:management"
+    ports:
+      - "15672:15672"
+      - "5672:5672"
+    tty: true

integration/microservices/e2e/broadcast-rmq.spec.ts

@@ -1,8 +1,8 @@
-import * as express from 'express';
-import * as request from 'supertest';
-import { Test } from '@nestjs/testing';
 import { INestApplication } from '@nestjs/common';
 import { Transport } from '@nestjs/microservices';
+import { Test } from '@nestjs/testing';
+import * as express from 'express';
+import * as request from 'supertest';
 import { RMQController } from '../src/rmq/rmq.controller';
 
 describe('RabbimMQ transport', () => {
@@ -17,16 +17,16 @@ describe('RabbimMQ transport', () => {
     server = express();
     app = module.createNestApplication(server);
     app.connectMicroservice({
-        transport: Transport.RMQ,
-        options: {
-          urls: [`amqp://admin:admin@localhost`],
-          queue: 'test',
-          queueOptions: { durable: false },
-        },
+      transport: Transport.RMQ,
+      options: {
+        urls: [`amqp://localhost:5672`],
+        queue: 'test',
+        queueOptions: { durable: false },
+      },
     });
     await app.startAllMicroservicesAsync();
     await app.init();
-});
+  });
 
   it(`/POST`, () => {
     return request(server)
@@ -54,11 +54,16 @@ describe('RabbimMQ transport', () => {
     return request(server)
       .post('/concurrent')
       .send([
-        [1, 2, 3, 4, 5],
-        [6, 7, 8, 9, 10],
-        [11, 12, 13, 14, 15],
-        [16, 17, 18, 19, 20],
-        [21, 22, 23, 24, 25],
+        Array.from({ length: 10 }, (v, k) => k + 1),
+        Array.from({ length: 10 }, (v, k) => k + 11),
+        Array.from({ length: 10 }, (v, k) => k + 21),
+        Array.from({ length: 10 }, (v, k) => k + 31),
+        Array.from({ length: 10 }, (v, k) => k + 41),
+        Array.from({ length: 10 }, (v, k) => k + 51),
+        Array.from({ length: 10 }, (v, k) => k + 61),
+        Array.from({ length: 10 }, (v, k) => k + 71),
+        Array.from({ length: 10 }, (v, k) => k + 81),
+        Array.from({ length: 10 }, (v, k) => k + 91),
       ])
       .expect(200, 'true');
   });

integration/microservices/e2e/sum-rmq.spec.ts

@@ -7,27 +7,25 @@ import {
 } from '@nestjs/common';
 import { ClientProxyFactory } from '@nestjs/microservices';
 import { Observable, throwError } from 'rxjs';
-import { catchError, tap } from 'rxjs/operators';
+import { catchError } from 'rxjs/operators';
 
 @Controller()
 export class DisconnectedClientController {
   @Post()
   call(@Body() options): Observable<number> {
     const client = ClientProxyFactory.create(options);
-    return client
-      .send<number, number[]>({ cmd: 'none' }, [1, 2, 3])
-      .pipe(
-        tap(
+    return client.send<number, number[]>({ cmd: 'none' }, [1, 2, 3]).pipe(
+      /*tap(
           console.log.bind(console, 'data'),
           console.error.bind(console, 'error'),
+        ),*/
+      catchError(({ code }) =>
+        throwError(
+          code === 'ECONNREFUSED' || code === 'CONN_ERR'
+            ? new RequestTimeoutException('ECONNREFUSED')
+            : new InternalServerErrorException(),
         ),
-        catchError(({ code }) =>
-          throwError(
-            code === 'ECONNREFUSED' || code === 'CONN_ERR'
-              ? new RequestTimeoutException('ECONNREFUSED')
-              : new InternalServerErrorException(),
-          ),
-        ),
-      );
+      ),
+    );
   }
 }

integration/microservices/src/disconnected.controller.ts

@@ -1,10 +1,9 @@
 import { Controller, Get } from '@nestjs/common';
 import {
-  Client,
-  MessagePattern,
   ClientProxy,
-  Transport,
   ClientProxyFactory,
+  MessagePattern,
+  Transport,
 } from '@nestjs/microservices';
 import { Observable } from 'rxjs';
 import { scan, take } from 'rxjs/operators';
@@ -15,13 +14,13 @@ export class RMQBroadcastController {
 
   constructor() {
     this.client = ClientProxyFactory.create({
-        transport: Transport.RMQ,
-        options: {
-          urls: [`amqp://admin:admin@localhost`],
-          queue: 'test',
-          queueOptions: { durable: false },
-        },
-      });
+      transport: Transport.RMQ,
+      options: {
+        urls: [`amqp://localhost:5672`],
+        queue: 'test_broadcast',
+        queueOptions: { durable: false },
+      },
+    });
   }
 
   @Get('broadcast')

integration/microservices/src/rmq/rmq-broadcast.controller.ts

@@ -1,12 +1,11 @@
-import { Controller, Get, Post, Body, Query, HttpCode } from '@nestjs/common';
+import { Body, Controller, HttpCode, Post, Query } from '@nestjs/common';
 import {
-  Client,
-  MessagePattern,
   ClientProxy,
-  Transport,
   ClientProxyFactory,
+  MessagePattern,
+  Transport,
 } from '@nestjs/microservices';
-import { Observable, of, from } from 'rxjs';
+import { from, Observable, of } from 'rxjs';
 import { scan } from 'rxjs/operators';
 
 @Controller()
@@ -15,13 +14,13 @@ export class RMQController {
 
   constructor() {
     this.client = ClientProxyFactory.create({
-        transport: Transport.RMQ,
-        options: {
-          urls: [`amqp://admin:admin@localhost`],
-          queue: 'test',
-          queueOptions: { durable: false },
-        },
-      });
+      transport: Transport.RMQ,
+      options: {
+        urls: [`amqp://localhost:5672`],
+        queue: 'test',
+        queueOptions: { durable: false },
+      },
+    });
   }
 
   @Post()

integration/microservices/src/rmq/rmq.controller.ts

@@ -4,5 +4,5 @@ export enum Transport {
   NATS,
   MQTT,
   GRPC,
-  RMQ
-}
+  RMQ,
+}

packages/common/enums/transport.enum.ts

@@ -59,42 +59,39 @@ export class ClientRMQ extends ClientProxy {
     this.client && this.client.close();
   }
 
-  public listen() {
+  public consumeChannel() {
     this.channel.addSetup(channel =>
       channel.consume(
         this.replyQueue,
-        msg => {
-          this.responseEmitter.emit(msg.properties.correlationId, msg);
-        },
+        msg => this.responseEmitter.emit(msg.properties.correlationId, msg),
         { noAck: true },
       ),
     );
   }
 
   public connect(): Promise<any> {
-    if (this.client && this.channel) {
+    if (this.client) {
       return this.connection;
     }
     this.client = this.createClient();
     this.handleError(this.client);
 
     const connect$ = this.connect$(this.client);
     this.connection = this.mergeDisconnectEvent(this.client, connect$)
-      .pipe(
-        switchMap(() => {
-          return new Promise(resolve =>
-            this.client.createChannel({
-              json: false,
-              setup: async channel => this.setupChannel(channel, resolve),
-            }),
-          );
-        }),
-        share(),
-      )
+      .pipe(switchMap(() => this.createChannel()), share())
       .toPromise();
     return this.connection;
   }
 
+  public createChannel(): Promise<void> {
+    return new Promise(resolve => {
+      this.channel = this.client.createChannel({
+        json: false,
+        setup: channel => this.setupChannel(channel, resolve),
+      });
+    });
+  }
+
   public createClient<T = any>(): T {
     return amqp.connect(this.urls) as T;
   }
@@ -114,13 +111,14 @@ export class ClientRMQ extends ClientProxy {
   public async setupChannel(channel: any, resolve: Function) {
     await channel.assertQueue(this.queue, this.queueOptions);
     await channel.prefetch(this.prefetchCount, this.isGlobalPrefetchCount);
+
     this.replyQueue = (await channel.assertQueue('', {
       exclusive: true,
     })).queue;
 
     this.responseEmitter = new EventEmitter();
     this.responseEmitter.setMaxListeners(0);
-    this.listen();
+    this.consumeChannel();
 
     resolve();
   }
@@ -131,10 +129,9 @@ export class ClientRMQ extends ClientProxy {
   ): Function {
     try {
       const correlationId = randomStringGenerator();
-      const listener = msg => {
-        const { content } = msg;
-        this.handleMessage(content, callback);
-      };
+      const listener = ({ content }) =>
+        this.handleMessage(JSON.parse(content.toString()), callback);
+
       this.responseEmitter.on(correlationId, listener);
       this.channel.sendToQueue(
         this.queue,

packages/microservices/client/client-rmq.ts

@@ -6,7 +6,7 @@ import { ServerMqtt } from './server-mqtt';
 import { ServerNats } from './server-nats';
 import { ServerRedis } from './server-redis';
 import { ServerTCP } from './server-tcp';
-import { ServerRMQ } from './server-rqm';
+import { ServerRMQ } from './server-rmq';
 
 export class ServerFactory {
   public static create(