feat(microservices) add capability to use RegExp in Kafka Events

closes nestjs#3083

Author: alfredoperez

packages/microservices/interfaces/message-handler.interface.ts

@@ -4,3 +4,8 @@ export interface MessageHandler<TInput = any, TContext = any, TResult = any> {
   (data: TInput, ctx?: TContext): Promise<Observable<TResult>>;
   isEventHandler?: boolean;
 }
+
+export interface RegExpMessageHandler {
+  pattern: RegExp;
+  messageHandler: MessageHandler;
+}

packages/microservices/server/server-kafka.ts

@@ -98,8 +98,11 @@ export class ServerKafka extends Server implements CustomTransportStrategy {
   }
 
   public async bindEvents(consumer: Consumer) {
-    const registeredPatterns = [...this.messageHandlers.keys()];
-    const subscribeToPattern = async (pattern: string) =>
+    const registeredPatterns = [
+      ...this.messageHandlers.keys(),
+      ...this.regExpMessageHandlers.map(handler => handler.pattern),
+    ];
+    const subscribeToPattern = async (pattern: string | RegExp) =>
       consumer.subscribe({
         topic: pattern,
       });

packages/microservices/server/server.ts

@@ -22,6 +22,7 @@ import {
   NatsOptions,
   ReadPacket,
   RedisOptions,
+  RegExpMessageHandler,
   RmqOptions,
   TcpOptions,
   WritePacket,
@@ -34,6 +35,7 @@ import { NO_EVENT_HANDLER } from '../constants';
 
 export abstract class Server {
   protected readonly messageHandlers = new Map<string, MessageHandler>();
+  protected readonly regExpMessageHandlers = new Array<RegExpMessageHandler>();
   protected readonly logger = new Logger(Server.name);
   protected serializer: ConsumerSerializer;
   protected deserializer: ConsumerDeserializer;
@@ -43,9 +45,13 @@ export abstract class Server {
     callback: MessageHandler,
     isEventHandler = false,
   ) {
-    const route = transformPatternToRoute(pattern);
-    callback.isEventHandler = isEventHandler;
-    this.messageHandlers.set(route, callback);
+    if (pattern.constructor.name === 'RegExp') {
+      this.regExpMessageHandlers.push({ pattern, messageHandler: callback });
+    } else {
+      const route = transformPatternToRoute(pattern);
+      callback.isEventHandler = isEventHandler;
+      this.messageHandlers.set(route, callback);
+    }
   }
 
   public getHandlers(): Map<string, MessageHandler> {
@@ -54,9 +60,20 @@ export abstract class Server {
 
   public getHandlerByPattern(pattern: string): MessageHandler | null {
     const route = this.getRouteFromPattern(pattern);
-    return this.messageHandlers.has(route)
-      ? this.messageHandlers.get(route)
-      : null;
+    let handler = null;
+    // Try to find the message handler by name
+    if (this.messageHandlers.has(route)) {
+      return this.messageHandlers.get(route);
+    }
+
+    // If it was not found, iterate through the Regular Expression handlers
+    this.regExpMessageHandlers.forEach(regExpHandler => {
+      if (regExpHandler.pattern.exec(route) !== null) {
+        handler = regExpHandler.messageHandler;
+      }
+    });
+
+    return handler;
   }
 
   public send(

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

@@ -3,6 +3,7 @@ import { Observable, of, throwError as _throw } from 'rxjs';
 import * as sinon from 'sinon';
 import { Server } from '../../server/server';
 import * as Utils from '../../utils';
+import { RegExpMessageHandler } from '../../interfaces';
 
 class TestServer extends Server {
   public listen(callback: () => void) {}
@@ -197,6 +198,27 @@ describe('Server', () => {
       });
     });
 
+    describe('when handler exists and was added with a RegExp', () => {
+      beforeEach(() => {
+        sandbox.stub(server as any, 'regExpMessageHandlers').value([
+          ({
+            pattern: /.*el.*/,
+            messageHandler: callback,
+          } as unknown) as RegExpMessageHandler,
+        ]);
+      });
+
+      it('should return expected handler', () => {
+        messageHandlersHasSpy.returns(false);
+
+        const value = server.getHandlerByPattern(handlerRoute);
+
+        expect(messageHandlersHasSpy.called).to.be.true;
+        expect(messageHandlersGetSpy.called).to.be.false;
+        expect(value).to.be.equal(callback);
+      });
+    });
+
     describe('when handler does not exists', () => {
       it('should return null', () => {
         messageHandlersHasSpy.returns(false);