Merge pull request #11642 from ryoctrl/nack_when_no_matching_handler

fix(microservices): to nack when there is no matching handler

Author: kamilmysliwiec

packages/microservices/constants.ts

@@ -40,6 +40,11 @@ export const RQM_DEFAULT_QUEUE_OPTIONS = {};
 export const RQM_DEFAULT_NOACK = true;
 export const RQM_DEFAULT_PERSISTENT = false;
 export const RQM_DEFAULT_NO_ASSERT = false;
+export const RQM_NO_EVENT_HANDLER = (
+  text: TemplateStringsArray,
+  pattern: string,
+) =>
+  `An unsupported event was received. It has been acknowledged, so it will not be re-delivered. Pattern: ${pattern}`;
 export const GRPC_DEFAULT_PROTO_LOADER = '@grpc/proto-loader';
 
 export const NO_EVENT_HANDLER = (text: TemplateStringsArray, pattern: string) =>

packages/microservices/server/server-rmq.ts

@@ -17,6 +17,7 @@ import {
   RQM_DEFAULT_QUEUE,
   RQM_DEFAULT_QUEUE_OPTIONS,
   RQM_DEFAULT_URL,
+  RQM_NO_EVENT_HANDLER,
 } from '../constants';
 import { RmqContext } from '../ctx-host';
 import { Transport } from '../enums';
@@ -25,6 +26,7 @@ import { CustomTransportStrategy, RmqOptions } from '../interfaces';
 import {
   IncomingRequest,
   OutgoingResponse,
+  ReadPacket,
 } from '../interfaces/packet.interface';
 import { RmqRecordSerializer } from '../serializers/rmq-record.serializer';
 import { Server } from './server';
@@ -42,6 +44,7 @@ export class ServerRMQ extends Server implements CustomTransportStrategy {
   protected readonly urls: string[] | RmqUrl[];
   protected readonly queue: string;
   protected readonly prefetchCount: number;
+  protected readonly noAck: boolean;
   protected readonly queueOptions: any;
   protected readonly isGlobalPrefetchCount: boolean;
   protected readonly noAssert: boolean;
@@ -54,6 +57,7 @@ export class ServerRMQ extends Server implements CustomTransportStrategy {
     this.prefetchCount =
       this.getOptionsProp(this.options, 'prefetchCount') ||
       RQM_DEFAULT_PREFETCH_COUNT;
+    this.noAck = this.getOptionsProp(this.options, 'noAck', RQM_DEFAULT_NOACK);
     this.isGlobalPrefetchCount =
       this.getOptionsProp(this.options, 'isGlobalPrefetchCount') ||
       RQM_DEFAULT_IS_GLOBAL_PREFETCH_COUNT;
@@ -141,8 +145,6 @@ export class ServerRMQ extends Server implements CustomTransportStrategy {
   }
 
   public async setupChannel(channel: any, callback: Function) {
-    const noAck = this.getOptionsProp(this.options, 'noAck', RQM_DEFAULT_NOACK);
-
     if (!this.queueOptions.noAssert) {
       await channel.assertQueue(this.queue, this.queueOptions);
     }
@@ -151,7 +153,7 @@ export class ServerRMQ extends Server implements CustomTransportStrategy {
       this.queue,
       (msg: Record<string, any>) => this.handleMessage(msg, channel),
       {
-        noAck,
+        noAck: this.noAck,
       },
     );
     callback();
@@ -200,6 +202,19 @@ export class ServerRMQ extends Server implements CustomTransportStrategy {
     response$ && this.send(response$, publish);
   }
 
+  public async handleEvent(
+    pattern: string,
+    packet: ReadPacket,
+    context: RmqContext,
+  ): Promise<any> {
+    const handler = this.getHandlerByPattern(pattern);
+    if (!handler && !this.noAck) {
+      this.channel.nack(context.getMessage(), false, false);
+      return this.logger.warn(RQM_NO_EVENT_HANDLER`${pattern}`);
+    }
+    return super.handleEvent(pattern, packet, context);
+  }
+
   public sendMessage<T = any>(
     message: T,
     replyTo: any,

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

@@ -3,6 +3,7 @@ import * as sinon from 'sinon';
 import { NO_MESSAGE_HANDLER } from '../../constants';
 import { BaseRpcContext } from '../../ctx-host/base-rpc.context';
 import { ServerRMQ } from '../../server/server-rmq';
+import { RmqContext } from '../../ctx-host';
 
 describe('ServerRMQ', () => {
   let server: ServerRMQ;
@@ -99,10 +100,15 @@ describe('ServerRMQ', () => {
       data: 'tests',
       id: '3',
     });
+    const channel = {
+      nack: sinon.spy(),
+    };
+
     let sendMessageStub: sinon.SinonStub;
 
     beforeEach(() => {
       sendMessageStub = sinon.stub(server, 'sendMessage').callsFake(() => ({}));
+      (server as any).channel = channel;
     });
     it('should call "handleEvent" if identifier is not present', async () => {
       const handleEventSpy = sinon.spy(server, 'handleEvent');
@@ -223,9 +229,33 @@ describe('ServerRMQ', () => {
       server.handleEvent(
         channel,
         { pattern: '', data },
-        new BaseRpcContext([]),
+        new RmqContext([{}, {}, '']),
       );
       expect(handler.calledWith(data)).to.be.true;
     });
+
+    it('should negative acknowledge without retrying if key does not exists in handlers object and noAck option is false', () => {
+      const nack = sinon.spy();
+      const message = { pattern: 'no-exists', data };
+      (server as any).channel = {
+        nack,
+      };
+      (server as any).noAck = false;
+      server.handleEvent(channel, message, new RmqContext([message, '', '']));
+
+      expect(nack.calledWith(message, false, false)).to.be.true;
+    });
+
+    it('should not negative acknowledge if key does not exists in handlers object but noAck option is true', () => {
+      const nack = sinon.spy();
+      const message = { pattern: 'no-exists', data };
+      (server as any).channel = {
+        nack,
+      };
+      (server as any).noAck = true;
+      server.handleEvent(channel, message, new RmqContext([message, '', '']));
+
+      expect(nack.calledWith(message, false, false)).not.to.be.true;
+    });
   });
 });