feat(microservices): implement preRequest message/event handler

integration/cors/e2e/fastify.spec.ts

@@ -147,7 +147,7 @@ describe('Fastify Cors', () => {
     after(async () => {
       await app.close();
     });
-    
+
     describe('Application Options', () => {
       before(async () => {
         const module = await Test.createTestingModule({

integration/nest-application/global-prefix/e2e/global-prefix.spec.ts

@@ -115,9 +115,7 @@ describe('Global prefix', () => {
       .get('/middleware/foo')
       .expect(200, MIDDLEWARE_PARAM_VALUE);
 
-    await request(server)
-      .get('/api/v1/middleware/foo')
-      .expect(404);
+    await request(server).get('/api/v1/middleware/foo').expect(404);
   });
 
   afterEach(async () => {

packages/common/interfaces/nest-application-context-options.interface.ts

@@ -29,4 +29,11 @@ export class NestApplicationContextOptions {
    * @default true
    */
   autoFlushLogs?: boolean;
+
+  /**
+   * @desc A token for a Class, that will be called per each message received
+   * by the microservice, before the actual handler.
+   * @default null;
+   */
+  preRequest?: any;
 }

packages/microservices/interfaces/microservice-configuration.interface.ts

@@ -33,7 +33,9 @@ export interface CustomStrategy {
 export interface GrpcOptions {
   transport?: Transport.GRPC;
   options: {
-    interceptors?: Array<(options: any, nextCall: (options: any) => any) => any>;
+    interceptors?: Array<
+      (options: any, nextCall: (options: any) => any) => any
+    >;
     url?: string;
     maxSendMessageLength?: number;
     maxReceiveMessageLength?: number;

packages/microservices/interfaces/pre-request-handler.ts

@@ -0,0 +1,5 @@
+import { BaseRpcContext } from '../ctx-host/base-rpc.context';
+
+export interface PreRequestHandler<TContext = BaseRpcContext, TData = any> {
+  handle(data: TData, context: TContext, handler: CallableFunction);
+}

packages/microservices/listeners-controller.ts

@@ -33,6 +33,7 @@ import {
 import { ListenerMetadataExplorer } from './listener-metadata-explorer';
 import { ServerGrpc } from './server';
 import { Server } from './server/server';
+import { PreRequestHandler } from './interfaces/pre-request-handler';
 
 export class ListenersController {
   private readonly metadataExplorer = new ListenerMetadataExplorer(
@@ -47,6 +48,7 @@ export class ListenersController {
     private readonly injector: Injector,
     private readonly clientFactory: IClientProxyFactory,
     private readonly exceptionFiltersContext: ExceptionFiltersContext,
+    private readonly preRequestHandler?: PreRequestHandler,
   ) {}
 
   public registerPatternHandlers(
@@ -98,11 +100,25 @@ export class ListenersController {
               );
               return originalReturnValue;
             };
+
+            if (this.preRequestHandler) {
+              const handler = (data, context) =>
+                this.preRequestHandler.handle(data, context, eventHandler);
+              return server.addHandler(pattern, handler, isEventHandler);
+            }
+
             return server.addHandler(pattern, eventHandler, isEventHandler);
           } else {
+            if (this.preRequestHandler) {
+              const handler = (data, context) =>
+                this.preRequestHandler.handle(data, context, proxy);
+              return server.addHandler(pattern, handler, isEventHandler);
+            }
+
             return server.addHandler(pattern, proxy, isEventHandler);
           }
         }
+
         const asyncHandler = this.createRequestScopedHandler(
           instanceWrapper,
           pattern,

packages/microservices/microservices-module.ts

@@ -15,15 +15,21 @@ import { ClientsContainer } from './container';
 import { ExceptionFiltersContext } from './context/exception-filters-context';
 import { RpcContextCreator } from './context/rpc-context-creator';
 import { RpcProxy } from './context/rpc-proxy';
-import { CustomTransportStrategy } from './interfaces';
+import { CustomTransportStrategy, MicroserviceOptions } from './interfaces';
 import { ListenersController } from './listeners-controller';
 import { Server } from './server/server';
+import { NestMicroserviceOptions } from '@nestjs/common/interfaces/microservices/nest-microservice-options.interface';
+import { PreRequestHandler } from './interfaces/pre-request-handler';
 
 export class MicroservicesModule {
   private readonly clientsContainer = new ClientsContainer();
   private listenersController: ListenersController;
 
-  public register(container: NestContainer, config: ApplicationConfig) {
+  public register(
+    container: NestContainer,
+    config: ApplicationConfig,
+    microservicesConfig: NestMicroserviceOptions & MicroserviceOptions,
+  ) {
     const rpcProxy = new RpcProxy();
     const exceptionFiltersContext = new ExceptionFiltersContext(
       container,
@@ -48,6 +54,7 @@ export class MicroservicesModule {
       injector,
       ClientProxyFactory,
       exceptionFiltersContext,
+      this.getPreRequestHandler(microservicesConfig?.preRequest, container),
     );
   }
 
@@ -103,4 +110,27 @@ export class MicroservicesModule {
     await Promise.all(clients.map(client => client.close()));
     this.clientsContainer.clear();
   }
+
+  private getPreRequestHandler(
+    preRequestToken,
+    container: NestContainer,
+  ): PreRequestHandler {
+    if (preRequestToken) {
+      const modules = container.getModules();
+
+      for (const module of modules.values()) {
+        const provider = module.getProviderByKey(preRequestToken);
+
+        if (provider) {
+          return provider.instance;
+        }
+      }
+
+      throw new RuntimeException(
+        'Cannot find preRequest instance! Please make sure that there is a corresponding provider for your token!',
+      );
+    }
+
+    return null;
+  }
 }

packages/microservices/nest-microservice.ts

@@ -46,7 +46,11 @@ export class NestMicroservice
   ) {
     super(container);
 
-    this.microservicesModule.register(container, this.applicationConfig);
+    this.microservicesModule.register(
+      container,
+      this.applicationConfig,
+      config,
+    );
     this.createServer(config);
     this.selectContextModule();
   }