fix(microservices): fix amqp v4 incompatibility issue

Author: kamilmysliwiec

packages/microservices/client/client-rmq.ts

@@ -1,10 +1,12 @@
 import { Logger } from '@nestjs/common/services/logger.service';
 import { loadPackage } from '@nestjs/common/utils/load-package.util';
 import { randomStringGenerator } from '@nestjs/common/utils/random-string-generator.util';
+import { isUndefined } from '@nestjs/common/utils/shared.utils';
 import { EventEmitter } from 'events';
 import { EmptyError, fromEvent, lastValueFrom, merge, Observable } from 'rxjs';
 import { first, map, share, switchMap } from 'rxjs/operators';
 import {
+  CONNECT_FAILED_EVENT,
   DISCONNECTED_RMQ_MESSAGE,
   DISCONNECT_EVENT,
   ERROR_EVENT,
@@ -110,7 +112,11 @@ export class ClientRMQ extends ClientProxy {
     instance: any,
     source$: Observable<T>,
   ): Observable<T> {
-    const close$ = fromEvent(instance, DISCONNECT_EVENT).pipe(
+    const connectFailedEvent = isUndefined(instance.connectionAttempts)
+      ? DISCONNECT_EVENT
+      : CONNECT_FAILED_EVENT;
+
+    const close$ = fromEvent(instance, connectFailedEvent).pipe(
       map((err: any) => {
         throw err;
       }),

packages/microservices/constants.ts

@@ -11,6 +11,7 @@ export const KAFKA_DEFAULT_BROKER = 'localhost:9092';
 
 export const CONNECT_EVENT = 'connect';
 export const DISCONNECT_EVENT = 'disconnect';
+export const CONNECT_FAILED_EVENT = 'connectFailed';
 export const MESSAGE_EVENT = 'message';
 export const DATA_EVENT = 'data';
 export const ERROR_EVENT = 'error';

packages/microservices/server/server-rmq.ts

@@ -3,7 +3,6 @@ import {
   isString,
   isUndefined,
 } from '@nestjs/common/utils/shared.utils';
-import { Observable } from 'rxjs';
 import {
   CONNECT_EVENT,
   DISCONNECTED_RMQ_MESSAGE,