Merge branch 'buxfix/3138-rabbitMQ-consumer-already-set' of https://github.com/sergey-telpuk/nest into sergey-telpuk-buxfix/3138-rabbitMQ-consumer-already-set

Author: kamilmysliwiec

packages/microservices/client/client-rmq.ts

@@ -74,6 +74,7 @@ export class ClientRMQ extends ClientProxy {
     }
     this.client = this.createClient();
     this.handleError(this.client);
+    this.handleDisconnectError(this.client);
 
     const connect$ = this.connect$(this.client);
     this.connection = this.mergeDisconnectEvent(this.client, connect$)
@@ -82,6 +83,7 @@ export class ClientRMQ extends ClientProxy {
         share(),
       )
       .toPromise();
+
     return this.connection;
   }
 
@@ -134,6 +136,14 @@ export class ClientRMQ extends ClientProxy {
     client.addListener(ERROR_EVENT, (err: any) => this.logger.error(err));
   }
 
+  public handleDisconnectError(client: any): void {
+    client.addListener(DISCONNECT_EVENT, (err: any) => {
+      this.close();
+      this.client = null;
+      return this.logger.error(err);
+    });
+  }
+
   public handleMessage(
     packet: unknown,
     callback: (packet: WritePacket) => any,

packages/microservices/server/server-rmq.ts

@@ -69,6 +69,9 @@ export class ServerRMQ extends Server implements CustomTransportStrategy {
   public async start(callback?: () => void) {
     this.server = this.createClient();
     this.server.on(CONNECT_EVENT, (_: any) => {
+      if (this.channel) {
+        return;
+      }
       this.channel = this.server.createChannel({
         json: false,
         setup: (channel: any) => this.setupChannel(channel, callback),