Merge branch 'sergey-telpuk-buxfix/3138-rabbitMQ-consumer-already-set'

Author: kamilmysliwiec

packages/microservices/client/client-rmq.ts

@@ -5,6 +5,7 @@ import { EventEmitter } from 'events';
 import { fromEvent, merge, Observable } from 'rxjs';
 import { first, map, share, switchMap } from 'rxjs/operators';
 import {
+  DISCONNECTED_RMQ_MESSAGE,
   DISCONNECT_EVENT,
   ERROR_EVENT,
   RQM_DEFAULT_IS_GLOBAL_PREFETCH_COUNT,
@@ -52,6 +53,8 @@ export class ClientRMQ extends ClientProxy {
   public close(): void {
     this.channel && this.channel.close();
     this.client && this.client.close();
+    this.channel = null;
+    this.client = null;
   }
 
   public consumeChannel() {
@@ -74,6 +77,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 +86,7 @@ export class ClientRMQ extends ClientProxy {
         share(),
       )
       .toPromise();
+
     return this.connection;
   }
 
@@ -134,6 +139,15 @@ 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.logger.error(DISCONNECTED_RMQ_MESSAGE);
+      this.logger.error(err);
+
+      this.close();
+    });
+  }
+
   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),