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

Author: yatin166

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,7 +135,7 @@ export class ServerRMQ extends Server<RmqEvents, RmqStatus> {
     this.pendingEventListeners = [];
 
     const connectFailedEvent = 'connectFailed';
-    this.server!.once(connectFailedEvent, (error: Record<string, unknown>) => {
+    this.server!.once(connectFailedEvent, async (error: Record<string, unknown>) => {
       this._status$.next(RmqStatus.DISCONNECTED);
 
       this.logger.error(CONNECTION_FAILED_MESSAGE);
@@ -150,7 +150,7 @@ export class ServerRMQ extends Server<RmqEvents, RmqStatus> {
         return;
       }
       if (++this.connectionAttempts === maxConnectionAttempts) {
-        this.close();
+        await this.close();
         callback?.(error.err ?? new Error(CONNECTION_FAILED_MESSAGE));
       }
     });

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

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