[messaging] feat: Introduce Messaging Lifecycle Hooks

README.md

@@ -219,6 +219,8 @@ flowchart LR
     MW --> H
 ```
 
+---
+
 ### Exception handling flow:
 
 ```mermaid

packages/messaging/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@nestjstools/messaging",
-  "version": "4.1.1",
+  "version": "4.2.0",
   "description": "Simplifies asynchronous and synchronous message handling with support for buses, handlers, channels, and consumers. Build scalable, decoupled applications with ease and reliability.",
   "author": "Sebastian Iwanczyszyn",
   "license": "MIT",

packages/messaging/src/bus/consumer.message-bus.ts

@@ -11,6 +11,9 @@ import { ConsumerDispatchedMessageError } from '../consumer/consumer-dispatched-
 import { HandlersException } from '../exception/handlers.exception';
 import { ExceptionListenerHandler } from '../exception-listener/exception-listener-handler';
 import { ExceptionContext } from '../exception-listener/exception-context';
+import { MessagingLifecycleHookHandler } from '../lifecycle-hook/messaging-lifecycle-hook-handler';
+import { HookMessage } from '../lifecycle-hook/messaging-lifecycle-hook-listener';
+import { MessageFactory } from '../message/message.factory';
 
 export class ConsumerMessageBus {
   constructor(
@@ -19,6 +22,7 @@ export class ConsumerMessageBus {
     private readonly logger: MessagingLogger,
     private readonly consumer: IMessagingConsumer<any>,
     private readonly exceptionListenerHandler: ExceptionListenerHandler,
+    private readonly messagingHookHandler: MessagingLifecycleHookHandler,
   ) {}
 
   async dispatch(consumerMessage: ConsumerMessage): Promise<void> {
@@ -46,6 +50,14 @@ export class ConsumerMessageBus {
         ),
       );
 
+      await this.messagingHookHandler.handleOnConsumerHandledMessage(
+        HookMessage.fromSealedRoutingMessage(
+          routingMessage,
+          this.channel.config.name,
+          this.channel.constructor.name,
+        ),
+      );
+
       await this.messageBus.dispatch(routingMessage);
     } catch (e) {
       await this.consumer.onError(
@@ -74,6 +86,14 @@ export class ConsumerMessageBus {
           consumerMessage.routingKey,
         ),
       );
+
+      await this.messagingHookHandler.handleOnFailedMessageConsumer(
+        HookMessage.fromConsumerMessage(
+          consumerMessage,
+          this.channel.config.name,
+          this.channel.constructor.name,
+        ),
+      );
     }
 
     return Promise.resolve();

packages/messaging/src/bus/distributed-message.bus.ts

@@ -5,12 +5,15 @@ import { MessageBusCollection } from './message-bus.collection';
 import { RoutingMessage } from '../message/routing-message';
 import { MessageFactory } from '../message/message.factory';
 import { NormalizerRegistry } from '../normalizer/normalizer.registry';
+import { MessagingLifecycleHookHandler } from '../lifecycle-hook/messaging-lifecycle-hook-handler';
+import { HookMessage } from '../lifecycle-hook/messaging-lifecycle-hook-listener';
 
 @Injectable()
 export class DistributedMessageBus implements IMessageBus {
   constructor(
     private messageBusCollection: MessageBusCollection,
     private normalizerRegistry: NormalizerRegistry,
+    private messagingLifecycleHookHandler: MessagingLifecycleHookHandler,
   ) {}
 
   async dispatch(message: RoutingMessage): Promise<MessageResponse> {
@@ -20,12 +23,30 @@ export class DistributedMessageBus implements IMessageBus {
 
     const response = [];
     for (const collection of this.messageBusCollection.getAll()) {
+      await this.messagingLifecycleHookHandler.handleBeforeMessageNormalization(
+        HookMessage.fromRoutingMessage(
+          message,
+          collection.channel.config.name,
+          collection.channel.constructor.name,
+        ),
+      );
+
       const normalizedMessage = await this.normalizerRegistry
         .getByName(collection.channel.config.normalizer.name)
         .normalize(message.message, message.messageRoutingKey);
+
+      await this.messagingLifecycleHookHandler.handleAfterMessageNormalization(
+        HookMessage.fromRoutingMessage(
+          message,
+          collection.channel.config.name,
+          collection.channel.constructor.name,
+        ),
+      );
+
       const handlerResponse = await collection.messageBus.dispatch(
         MessageFactory.creteSealedFromMessage(normalizedMessage, message),
       );
+
       if (handlerResponse) {
         response.push(handlerResponse);
       }

packages/messaging/src/bus/in-memory-message-bus.factory.ts

@@ -8,6 +8,7 @@ import { MessageBusFactory } from '../dependency-injection/decorator';
 import { InMemoryMessageBus } from './in-memory-message.bus';
 import { IMessageBusFactory } from './i-message-bus.factory';
 import { NormalizerRegistry } from '../normalizer/normalizer.registry';
+import { MessagingLifecycleHookHandler } from '../lifecycle-hook/messaging-lifecycle-hook-handler';
 
 @Injectable()
 @MessageBusFactory(InMemoryChannel)
@@ -19,6 +20,7 @@ export class InMemoryMessageBusFactory implements IMessageBusFactory<InMemoryCha
     private middlewareRegistry: MiddlewareRegistry,
     @Inject(Service.MESSAGE_NORMALIZERS_REGISTRY)
     private messageNormalizerRegistry: NormalizerRegistry,
+    private messagingHookHandler: MessagingLifecycleHookHandler,
   ) {}
 
   create(channel: InMemoryChannel): IMessageBus {
@@ -27,6 +29,7 @@ export class InMemoryMessageBusFactory implements IMessageBusFactory<InMemoryCha
       this.middlewareRegistry,
       channel,
       this.messageNormalizerRegistry,
+      this.messagingHookHandler,
     );
   }
 }

packages/messaging/src/bus/in-memory-message.bus.ts

@@ -13,13 +13,16 @@ import { Message } from '../message/message';
 import { RoutingMessage } from '../message/routing-message';
 import { NormalizerRegistry } from '../normalizer/normalizer.registry';
 import { DefaultMessageOptions } from '../message/default-message-options';
+import { MessagingLifecycleHookHandler } from '../lifecycle-hook/messaging-lifecycle-hook-handler';
+import { HookMessage } from '../lifecycle-hook/messaging-lifecycle-hook-listener';
 
 export class InMemoryMessageBus implements IMessageBus {
   constructor(
     private registry: MessageHandlerRegistry,
     private middlewareRegistry: MiddlewareRegistry,
     private channel: InMemoryChannel,
     private normalizerRegistry: NormalizerRegistry,
+    private messagingHookHandler: MessagingLifecycleHookHandler,
   ) {}
 
   async dispatch(message: Message): Promise<object | void> {
@@ -30,6 +33,28 @@ export class InMemoryMessageBus implements IMessageBus {
       HandlerMiddleware,
     );
 
+    let messageToDispatch =
+      message instanceof RoutingMessage ? message.message : {};
+
+    if (message instanceof SealedRoutingMessage) {
+      // Sealed messages carry raw payload and must be denormalized before dispatch.
+      const normalizerDefinition: object =
+        message.messageOptions instanceof DefaultMessageOptions
+          ? message.messageOptions.normalizer
+          : ObjectForwardMessageNormalizer;
+
+      messageToDispatch = await this.normalizerRegistry
+        .getByName(normalizerDefinition['name'])
+        .denormalize(message.message, message.messageRoutingKey);
+    }
+
+    // Hook fired once the payload shape is ready for handler pipeline.
+    await this.messagingHookHandler.handleAfterMessageDenormalized(
+      HookMessage.fromRoutingMessage(
+        MessageFactory.creteRoutingFromMessage(messageToDispatch, message),
+      ),
+    );
+
     try {
       this.registry.getByRoutingKey(message.messageRoutingKey);
     } catch (e) {
@@ -48,6 +73,7 @@ export class InMemoryMessageBus implements IMessageBus {
           avoidErrorsForNonExistedHandlers;
       }
 
+      // Missing handler can be configured as no-op for fire-and-forget scenarios.
       if (avoidErrorsForNonExistedHandlers) {
         return Promise.resolve();
       }
@@ -60,27 +86,27 @@ export class InMemoryMessageBus implements IMessageBus {
         DecoratorExtractor.extractMessageMiddleware(middleware),
       ),
     );
-    const context = MiddlewareContext.createFresh(middlewareInstances);
 
-    let messageToDispatch =
-      message instanceof RoutingMessage ? message.message : {};
-
-    if (message instanceof SealedRoutingMessage) {
-      const normalizerDefinition: object =
-        message.messageOptions instanceof DefaultMessageOptions
-          ? message.messageOptions.normalizer
-          : ObjectForwardMessageNormalizer;
+    const context = MiddlewareContext.createFresh(middlewareInstances);
 
-      messageToDispatch = await this.normalizerRegistry
-        .getByName(normalizerDefinition['name'])
-        .denormalize(message.message, message.messageRoutingKey);
-    }
+    // Hook around handler execution.
+    await this.messagingHookHandler.handleBeforeMessageHandler(
+      HookMessage.fromRoutingMessage(
+        MessageFactory.creteRoutingFromMessage(messageToDispatch, message),
+      ),
+    );
 
     const response = await middlewareInstances[0].process(
       MessageFactory.creteRoutingFromMessage(messageToDispatch, message),
       context,
     );
 
+    await this.messagingHookHandler.handleAfterMessageHandlerExecuted(
+      HookMessage.fromRoutingMessage(
+        MessageFactory.creteRoutingFromMessage(messageToDispatch, message),
+      ),
+    );
+
     return Promise.resolve(response);
   }
 }

packages/messaging/src/consumer/distributed.consumer.ts

@@ -1,4 +1,4 @@
-import { Inject } from '@nestjs/common';
+import { Inject, Injectable } from '@nestjs/common';
 import { DiscoveryService } from '@nestjs/core';
 import { Service } from '../dependency-injection/service';
 import { IMessageBus } from '../bus/i-message-bus';
@@ -9,7 +9,9 @@ import { MESSAGE_CONSUMER_METADATA } from '../dependency-injection/decorator';
 import { IMessagingConsumer } from './i-messaging-consumer';
 import { ExceptionListenerHandler } from '../exception-listener/exception-listener-handler';
 import { ConsumerMessageBus } from '../bus/consumer.message-bus';
+import { MessagingLifecycleHookHandler } from '../lifecycle-hook/messaging-lifecycle-hook-handler';
 
+@Injectable()
 export class DistributedConsumer {
   constructor(
     @Inject(Service.DEFAULT_MESSAGE_BUS)
@@ -20,6 +22,7 @@ export class DistributedConsumer {
     private readonly exceptionListenerHandler: ExceptionListenerHandler,
     @Inject(Service.LOGGER) private readonly logger: MessagingLogger,
     private readonly discoveryService: DiscoveryService,
+    private readonly messagingLifecycleHookHandler: MessagingLifecycleHookHandler,
   ) {}
 
   async run(): Promise<void> {
@@ -63,6 +66,7 @@ export class DistributedConsumer {
         this.logger,
         consumer,
         this.exceptionListenerHandler,
+        this.messagingLifecycleHookHandler,
       );
 
       await consumer.consume(dispatcher, channel);

packages/messaging/src/dependency-injection/decorator.ts

@@ -1,4 +1,5 @@
 import { ChannelConfig } from '../config';
+import { LifecycleHook } from '../lifecycle-hook/messaging-lifecycle-hook-listener';
 
 export const MESSAGE_HANDLER_METADATA = 'MESSAGE_HANDLER_METADATA';
 export const CHANNEL_FACTORY_METADATA = 'CHANNEL_FACTORY_METADATA';
@@ -9,6 +10,8 @@ export const MESSAGING_NORMALIZER_METADATA = 'MESSAGING_NORMALIZER_METADATA';
 export const MESSAGING_EXCEPTION_LISTENER_METADATA =
   'MESSAGING_EXCEPTION_LISTENER_METADATA';
 export const MESSAGING_MESSAGE_METADATA = 'MESSAGING_MESSAGE_METADATA';
+export const MESSAGING_LIFECYCLE_HOOK_METADATA =
+  'MESSAGING_LIFECYCLE_HOOK_METADATA';
 
 export const MessageHandler = (...routingKey: string[]): ClassDecorator => {
   return (target) => {
@@ -70,6 +73,18 @@ export const MessagingExceptionListener = (): ClassDecorator => {
   };
 };
 
+export const MessagingLifecycleHook = (
+  lifecycleHook: LifecycleHook,
+): ClassDecorator => {
+  return (target) => {
+    Reflect.defineMetadata(
+      MESSAGING_LIFECYCLE_HOOK_METADATA,
+      `${lifecycleHook}:${target.name}`,
+      target,
+    );
+  };
+};
+
 export function DenormalizeMessage(): ParameterDecorator {
   return (target, propertyKey, parameterIndex) => {
     const paramTypes = Reflect.getMetadata(

packages/messaging/src/dependency-injection/register.ts

@@ -5,6 +5,7 @@ import { Service } from './service';
 import {
   MESSAGE_HANDLER_METADATA,
   MESSAGING_EXCEPTION_LISTENER_METADATA,
+  MESSAGING_LIFECYCLE_HOOK_METADATA,
   MESSAGING_MIDDLEWARE_METADATA,
   MESSAGING_NORMALIZER_METADATA,
 } from './decorator';
@@ -13,6 +14,7 @@ import { Registry } from '../shared/registry';
 import { MiddlewareRegistry } from '../middleware/middleware.registry';
 import { ExceptionListenerRegistry } from '../exception-listener/exception-listener.registry';
 import { NormalizerRegistry } from '../normalizer/normalizer.registry';
+import { MessagingLifecycleHookRegistry } from '../lifecycle-hook/messaging-lifecycle-hook.registry';
 
 export const registerHandlers = (
   moduleRef: ModuleRef,
@@ -78,36 +80,47 @@ export const registerExceptionListener = (
   );
 };
 
+export const registerMessagingHooks = (
+  moduleRef: ModuleRef,
+  discoveryService: DiscoveryService,
+) => {
+  register<MessagingLifecycleHookRegistry>(
+    moduleRef,
+    discoveryService,
+    MessagingLifecycleHookRegistry,
+    MESSAGING_LIFECYCLE_HOOK_METADATA,
+    'MessagingLifecycleHook',
+  );
+};
+
 const register = <T extends Registry<object>>(
   moduleRef: ModuleRef,
   discoveryService: DiscoveryService,
-  registryProvider: string,
+  registryProvider:
+    | string
+    | symbol
+    | (abstract new (...args: unknown[]) => unknown),
   decoratorMetadata: string,
   name: string,
 ) => {
   const exceptions = [DEFAULT_NORMALIZER, DEFAULT_MIDDLEWARE];
   const registry: Registry<T> = moduleRef.get(registryProvider);
   const logger: MessagingLogger = moduleRef.get(Service.LOGGER);
-  const instances = discoveryService
-    .getProviders()
-    .filter((messageExceptionListener) => {
-      if (!messageExceptionListener.metatype) {
-        return false;
-      }
+  const instances = discoveryService.getProviders().filter((provider) => {
+    if (!provider.metatype) {
+      return false;
+    }
 
-      return Reflect.hasMetadata(
-        decoratorMetadata,
-        messageExceptionListener.metatype,
-      );
-    });
+    return Reflect.hasMetadata(decoratorMetadata, provider.metatype);
+  });
 
-  instances.forEach((messageExceptionListener) => {
+  instances.forEach((provider) => {
     registry.register(
-      Reflect.getMetadata(decoratorMetadata, messageExceptionListener.metatype),
-      messageExceptionListener.instance,
+      Reflect.getMetadata(decoratorMetadata, provider.metatype),
+      provider.instance,
     );
-    if (!exceptions.includes(messageExceptionListener.name)) {
-      logger.log(`${name} [${messageExceptionListener.name}] was registered`);
+    if (!exceptions.includes(provider.name)) {
+      logger.log(`${name} [${provider.name}] was registered`);
     }
   });
 };

packages/messaging/src/index.ts

@@ -31,3 +31,4 @@ export * from './exception/handlers.exception';
 export * from './logger/log';
 export * from './logger/messaging-logger';
 export * from './logger/nest-logger';
+export * from './lifecycle-hook/messaging-lifecycle-hook-listener';