import { Message } from '@node-ts/bus-messages' ; import { DefaultHandlerRegistry , Handler } from '../handler' ; import { HandlerDefinition , isClassHandler } from '../handler/handler' ; import { JsonSerializer , Serializer } from '../serialization' ; import { MemoryQueue , Transport , TransportMessage } from '../transport' ; import { ClassConstructor , CoreDependencies , Middleware , MiddlewareDispatcher } from '../util' ; import { BusInstance } from './bus-instance' ; import { Persistence , Workflow , WorkflowState } from '../workflow' ; import { WorkflowRegistry } from '../workflow/registry/workflow-registry' ; import { BusAlreadyInitialized } from './error' ; import { ContainerAdapter } from '../container' ; import { defaultLoggerFactory , LoggerFactory } from '../logger' ; import { ContainerNotRegistered } from '../error' ; import { MessageSerializer } from '../serialization/message-serializer' ; import { InMemoryPersistence } from '../workflow/persistence' ; import { DefaultRetryStrategy , RetryStrategy } from '../retry-strategy' ; export interface BusInitializeOptions { default sendOnly : boolean ; } const defaultBusInitializeOptions = { sendOnly : false } ; export class BusConfiguration { private configuredTransport : Transport | undefined ; private concurrency = 1 ; private busInstance : BusInstance | undefined ; private container : ContainerAdapter | undefined ; private workflowRegistry = new WorkflowRegistry ( ) ; private handlerRegistry = new DefaultHandlerRegistry ( ) ; private loggerFactory : LoggerFactory = defaultLoggerFactory ; private serializer = new JsonSerializer ( ) ; private persistence : Persistence = new InMemoryPersistence ( ) ; private messageReadMiddlewares = new MiddlewareDispatcher < TransportMessage < unknown > > ( ) ; private retryStrategy : RetryStrategy = new DefaultRetryStrategy ( ) ; param options async initialize ( options = defaultBusInitializeOptions ) { const { sendOnly } = options ; const logger = this . loggerFactory ( '@node-ts/bus-core:bus' ) ; logger . debug ( 'Initializing bus' , { sendOnly } ) ; if ( ! ! this . busInstance ) { throw new BusAlreadyInitialized ( ) ; } const coreDependencies = { container : this . container , handlerRegistry : this . handlerRegistry , loggerFactory : this . loggerFactory , serializer : this . serializer , messageSerializer : new MessageSerializer ( this . serializer , this . handlerRegistry ) , retryStrategy : this . retryStrategy } ; if ( ! sendOnly ) { this . persistence ?. prepare ( coreDependencies ) ; this . workflowRegistry . prepare ( coreDependencies , this . persistence ) ; await this . workflowRegistry . initialize ( this . handlerRegistry , this . container ) ; const classHandlers = this . handlerRegistry . getClassHandlers ( ) ; if ( ! this . container && classHandlers . length ) { throw new ContainerNotRegistered ( classHandlers [ 0 ] . constructor . name ) ; } } const transport = this . configuredTransport || new MemoryQueue ( ) ; transport . prepare ( coreDependencies ) ; if ( transport . connect ) { await transport . connect ( ) ; } if ( ! sendOnly && transport . initialize ) { await transport . initialize ( this . handlerRegistry ) ; } this . busInstance = new BusInstance ( transport , this . concurrency , this . workflowRegistry , coreDependencies , this . messageReadMiddlewares ) ; logger . debug ( 'Bus initialized' , { sendOnly , registeredMessages : this . handlerRegistry . getMessageNames ( ) } ) ; return this . busInstance ; } param messageType param messageHandler param customResolver withHandler ( classHandler ) ; withHandler < MessageType extends ( Message | object ) > ( functionHandler ) ; withHandler < MessageType extends ( Message | object ) > ( handler ) { if ( ! ! this . busInstance ) { throw new BusAlreadyInitialized ( ) ; } if ( 'messageHandler' in handler ) { this . handlerRegistry . register ( handler . messageType , handler . messageHandler ) ; } else if ( isClassHandler ( handler ) ) { const handlerInstance = new handler ( ) ; this . handlerRegistry . register ( handlerInstance . messageType , handler ) ; } return this ; } param messageHandler param customResolver withCustomHandler < MessageType extends ( Message | object ) > ( messageHandler , customResolver ) { if ( ! ! this . busInstance ) { throw new BusAlreadyInitialized ( ) ; } this . handlerRegistry . registerCustom ( messageHandler , customResolver ) ; return this ; } /**
     * Register a workflow definition so that all of the messages it depends on will be subscribed to
     * and forwarded to the handlers inside the workflow
     */ withWorkflow < TWorkflowState extends WorkflowState > ( workflow ) { if ( ! ! this . busInstance ) { throw new BusAlreadyInitialized ( ) ; } this . workflowRegistry . register ( workflow ) ; return this ; } /**
     * Configures Bus to use a different transport than the default MemoryQueue
     */ withTransport ( transportConfiguration ) { if ( ! ! this . busInstance ) { throw new BusAlreadyInitialized ( ) ; } this . configuredTransport = transportConfiguration ; return this ; } /**
     * Configures Bus to use a different logging provider than the default console logger
     */ withLogger ( loggerFactory ) { if ( ! ! this . busInstance ) { throw new BusAlreadyInitialized ( ) ; } this . loggerFactory = loggerFactory ; return this ; } /**
     * Configures Bus to use a different serialization provider. The provider is responsible for
     * transforming messages to/from a serialized representation, as well as ensuring all object
     * properties are a strong type
     */ withSerializer ( serializer ) { if ( ! ! this . busInstance ) { throw new BusAlreadyInitialized ( ) ; } this . serializer = serializer ; return this ; } /**
     * Configures Bus to use a different persistence provider than the default InMemoryPersistence provider.
     * This is used to persist workflow data and is unused if not using workflows.
     */ withPersistence ( persistence ) { if ( ! ! this . busInstance ) { throw new BusAlreadyInitialized ( ) ; } this . persistence = persistence ; return this ; } /**
     * Sets the message handling concurrency beyond the default value of 1, which will increase the number of messages
     * handled in parallel.
     */ withConcurrency ( concurrency ) { if ( concurrency < 1 ) { throw new Error ( 'Invalid concurrency setting. Must be set to 1 or greater' ) ; } this . concurrency = concurrency ; return this ; } param container withContainer ( container ) { this . container = container ; return this ; } /**
     * Register optional middlewares that will run for each message that is polled from the transport
     * Note these middlewares only run when polling successfully pulls a message off the Transports queue
     * After all the user defined middlewares have registered.
     */ withMessageReadMiddleware < TransportMessageType = unknown > ( messageReadMiddleware ) { this . messageReadMiddlewares . use ( messageReadMiddleware ) ; return this ; } default withRetryStrategy ( retryStrategy ) { this . retryStrategy = retryStrategy ; return this ; } }