Merge pull request #15032 from yatin166/fix/close-amqp-connection-properly

fix(microservices): ensure all the amqp connections are closed properly

Author: kamilmysliwiec

packages/microservices/server/server-rmq.ts

@@ -100,9 +100,9 @@ export class ServerRMQ extends Server<RmqEvents, RmqStatus> {
     }
   }
 
-  public close(): void {
-    this.channel && this.channel.close();
-    this.server && this.server.close();
+  public async close(): Promise<void> {
+    this.channel && (await this.channel.close());
+    this.server && (await this.server.close());
     this.pendingEventListeners = [];
   }
 
@@ -135,25 +135,28 @@ export class ServerRMQ extends Server<RmqEvents, RmqStatus> {
     this.pendingEventListeners = [];
 
     const connectFailedEvent = 'connectFailed';
-    this.server!.once(connectFailedEvent, (error: Record<string, unknown>) => {
-      this._status$.next(RmqStatus.DISCONNECTED);
-
-      this.logger.error(CONNECTION_FAILED_MESSAGE);
-      if (error?.err) {
-        this.logger.error(error.err);
-      }
-      const isReconnecting = !!this.channel;
-      if (
-        maxConnectionAttempts === INFINITE_CONNECTION_ATTEMPTS ||
-        isReconnecting
-      ) {
-        return;
-      }
-      if (++this.connectionAttempts === maxConnectionAttempts) {
-        this.close();
-        callback?.(error.err ?? new Error(CONNECTION_FAILED_MESSAGE));
-      }
-    });
+    this.server!.once(
+      connectFailedEvent,
+      async (error: Record<string, unknown>) => {
+        this._status$.next(RmqStatus.DISCONNECTED);
+
+        this.logger.error(CONNECTION_FAILED_MESSAGE);
+        if (error?.err) {
+          this.logger.error(error.err);
+        }
+        const isReconnecting = !!this.channel;
+        if (
+          maxConnectionAttempts === INFINITE_CONNECTION_ATTEMPTS ||
+          isReconnecting
+        ) {
+          return;
+        }
+        if (++this.connectionAttempts === maxConnectionAttempts) {
+          await this.close();
+          callback?.(error.err ?? new Error(CONNECTION_FAILED_MESSAGE));
+        }
+      },
+    );
   }
 
   public createClient<T = any>(): T {

packages/microservices/test/server/server-rmq.spec.ts

@@ -78,12 +78,12 @@ describe('ServerRMQ', () => {
       untypedServer.server = rmqServer;
       untypedServer.channel = rmqChannel;
     });
-    it('should close server', () => {
-      server.close();
+    it('should close server', async () => {
+      await server.close();
       expect(rmqServer.close.called).to.be.true;
     });
-    it('should close channel', () => {
-      server.close();
+    it('should close channel', async () => {
+      await server.close();
       expect(rmqChannel.close.called).to.be.true;
     });
   });