Adding context on multi consumer prehandler barrier (#34)

* Adding context to barrier callback for multi consumers

* Adjusting multiconsumers

* Fixing error on AbstractSnsSqsConsumerMonoSchema

packages/amqp/lib/AbstractAmqpConsumerMultiSchema.ts

@@ -53,8 +53,12 @@ export abstract class AbstractAmqpConsumerMultiSchema<
     return handler.messageLogFormatter(message)
   }
 
-  override preHandlerBarrier(message: MessagePayloadType, messageType: string): Promise<boolean> {
+  protected override async preHandlerBarrier(
+    message: MessagePayloadType,
+    messageType: string,
+  ): Promise<boolean> {
     const handler = this.handlerContainer.resolveHandler(messageType)
-    return handler.preHandlerBarrier ? handler.preHandlerBarrier(message) : Promise.resolve(true)
+    // @ts-ignore
+    return handler.preHandlerBarrier ? await handler.preHandlerBarrier(message, this) : true
   }
 }

packages/core/index.ts

@@ -33,7 +33,7 @@ export {
   HandlerContainer,
   MessageHandlerConfig,
   MessageHandlerConfigBuilder,
-  BarrierCallbackWithoutMessageType,
+  BarrierCallbackMultiConsumers,
 } from './lib/queues/HandlerContainer'
 export type { HandlerContainerOptions, Handler } from './lib/queues/HandlerContainer'
 

packages/core/lib/queues/HandlerContainer.ts

@@ -3,15 +3,16 @@ import type { ZodSchema } from 'zod'
 
 export type LogFormatter<MessagePayloadSchema> = (message: MessagePayloadSchema) => unknown
 
-export type BarrierCallbackWithoutMessageType<MessagePayloadSchema extends object> = (
+export type BarrierCallbackMultiConsumers<MessagePayloadSchema extends object, ExecutionContext> = (
   message: MessagePayloadSchema,
+  context: ExecutionContext,
 ) => Promise<boolean>
 
 export const defaultLogFormatter = <MessagePayloadSchema>(message: MessagePayloadSchema) => message
 
-export type HandlerConfigOptions<MessagePayloadSchema extends object> = {
+export type HandlerConfigOptions<MessagePayloadSchema extends object, ExecutionContext> = {
   messageLogFormatter?: LogFormatter<MessagePayloadSchema>
-  preHandlerBarrier?: BarrierCallbackWithoutMessageType<MessagePayloadSchema>
+  preHandlerBarrier?: BarrierCallbackMultiConsumers<MessagePayloadSchema, ExecutionContext>
 }
 
 export class MessageHandlerConfig<
@@ -21,12 +22,15 @@ export class MessageHandlerConfig<
   public readonly schema: ZodSchema<MessagePayloadSchema>
   public readonly handler: Handler<MessagePayloadSchema, ExecutionContext>
   public readonly messageLogFormatter: LogFormatter<MessagePayloadSchema>
-  public readonly preHandlerBarrier?: BarrierCallbackWithoutMessageType<MessagePayloadSchema>
+  public readonly preHandlerBarrier?: BarrierCallbackMultiConsumers<
+    MessagePayloadSchema,
+    ExecutionContext
+  >
 
   constructor(
     schema: ZodSchema<MessagePayloadSchema>,
     handler: Handler<MessagePayloadSchema, ExecutionContext>,
-    options?: HandlerConfigOptions<MessagePayloadSchema>,
+    options?: HandlerConfigOptions<MessagePayloadSchema, ExecutionContext>,
   ) {
     this.schema = schema
     this.handler = handler
@@ -45,7 +49,7 @@ export class MessageHandlerConfigBuilder<MessagePayloadSchemas extends object, E
   addConfig<MessagePayloadSchema extends MessagePayloadSchemas>(
     schema: ZodSchema<MessagePayloadSchema>,
     handler: Handler<MessagePayloadSchema, ExecutionContext>,
-    options?: HandlerConfigOptions<MessagePayloadSchema>,
+    options?: HandlerConfigOptions<MessagePayloadSchema, ExecutionContext>,
   ) {
     // @ts-ignore
     this.configs.push(new MessageHandlerConfig(schema, handler, options))

packages/sns/lib/sns/AbstractSnsSqsConsumerMonoSchema.ts

@@ -93,6 +93,16 @@ export abstract class AbstractSnsSqsConsumerMonoSchema<
     return readSnsMessage(message, this.errorResolver)
   }
 
+  /**
+   * Override to implement barrier pattern
+   */
+  protected preHandlerBarrier(
+    _message: MessagePayloadType,
+    _messageType: string,
+  ): Promise<boolean> {
+    return Promise.resolve(true)
+  }
+
   override async init(): Promise<void> {
     if (this.deletionConfig && this.creationConfig && this.subscriptionConfig) {
       await deleteSnsSqs(

packages/sqs/lib/sqs/AbstractSqsConsumerMultiSchema.ts

@@ -87,6 +87,7 @@ export abstract class AbstractSqsConsumerMultiSchema<
     messageType: string,
   ): Promise<boolean> {
     const handler = this.handlerContainer.resolveHandler(messageType)
-    return handler.preHandlerBarrier ? await handler.preHandlerBarrier(message) : true
+    // @ts-ignore
+    return handler.preHandlerBarrier ? await handler.preHandlerBarrier(message, this) : true
   }
 }