# @node-ts/bus A node library for building message driven distributed applications. {% embed url="" %} {% endembed %} Messaging is an old technology that's famous for reliability, resiliency and throughput. Message based systems can scale easily and suffer outages without loss of data, and can self-recover once outages are resolved. **@node-ts/bus** is a message bus library that makes it easier to build message based systems in node. It takes care of configuring an underlying transport (eg RabbitMQ), routing messages to handlers, propagating attributes etc. By abstracting away the technical complexities of working with message systems, more of your codebase remains dedicated to the concerns of your application. ### Resiliency Message based systems can suffer environmental instability in a way where data is not lost. Environmental instability can be caused through network partitioning, system restarts, data corruption, bugs, security misconfigurations etc, and can all lead to part or all of your system being unavailable. During this type if your application is still processing user input, it may be throwing errors, corrupting the state of the data and discarding requests as a result. Once the system is back online, it can take a lot of time and effort to restore the state to an uncorrupted form if at all. Message systems don't suffer from these issues. When part of the system is down a message will attempt to be processed and will fail. Processing of the message will automatically retry a number of times at which point the message will be routed to a dead letter queue. When the system comes back online, teams can replay messages in the dead letter queue to bring the system back into a valid state. ### Scalability **@node-ts/bus** helps improve scalability by ensuring services don't get overloaded. Since operations are pull-based your application will only fetch the next piece of work when it is ready to do so, and when overloaded the number of messages in the queue will grow. This compares to HTTP/REST based services which may start to drop requests when they become overloaded. This can make rules around autoscaling easier since the number of instances of your services can increase based on metrics such as the number of messages waiting in the queue, or the age of the oldest message. These metrics are more accurate than scaling on cpu, memory, response time etc which aren't an accurate representation of load and scale. # Getting started # Installation Install the npm package into your application. ``` npm i @node-ts/bus-core @node-ts/bus-messages --save ``` Configure and initialize the bus when your application starts up. ```typescript import { Bus } from '@node-ts/bus-core' ​ async function run () { const bus = await Bus.configure().initialize() } ``` This is the most basic of setups and your app is running an in memory queue capable of receiving messages from itself. In production, it's strongly recommended to [configure a transport](broken://pages/-MkbAMS7a-WL5iUiT-q6) so that your application can be distributed and survive restarts. # Handling messages At some point you'll want to handle messages that have been sent to your application. This can be done by creating a handler and registering it with the [BusConfiguration](/reference/busconfiguration). The following example creates a handler for a hotel booking application Declare a command that models the **ReserveRoom** command ```typescript // reserve-room.ts import { Command } from '@node-ts/bus-messages' export class ReserveRoom extends Command { $name = 'reservations/reserve-room' $version = 0 constructor ( readonly roomId: string, readonly bookingId: string ) { super() } } ``` Create a handler that receives a command to `ReserveRoom` that it delegates the operation to a `reservationService`. {% hint style="info" %} Handlers should be kept as dumb as possible and delegate the work to dedicated services. This will help keep the messaging concerns of your application decoupled from the actual work it needs to do. {% endhint %} ```typescript // reserve-room-handler.ts import { handlerFor } from '@node-ts/bus-core' import { ReserveRoom } from '../messages' import { reservationService } from '../services' export const reserveRoomHandler = handlerFor( ReserveRoom, command => reservationService.reserveRoom(command.reservation, command.bookingId) ) ``` Register the handler with the BusConfiguration. ```typescript // application.ts import { Bus, BusInstance } from '@node-ts/bus-core' import { reserveRoomHandler } from './handlers' import { ReserveRoom } from './messages' let bus: BusInstance const start = async () => { bus = await Bus.configure() .withHandler(reserveRoomHandler) .initialize() // Start the bus to commence processing messages await bus.start() } start .then(async () => bus.send(new ReserveRoom( '63a65cf0-d239-4b83-96da-f33f013db23a', '12b85a56-e929-47a8-9ac3-e87739d5d215' )) .catch(console.error) ``` # Shutting down cleanly Ensure your app shuts down cleanly when terminated Often you'll want to run your app in a way that it can terminate cleanly and give it time to finish doing its work before exiting. This is common if you manually terminate it (CTRL + C), send it a kill signal like `kill -INT 1234` or if the underlying pod/container/host is stopping. In these situations your app should finish processing the any messages its read from the queue and not read any more, allowing it to exit gracefully. To do this, hooking into the process signals and calling **dispose** on the bus is the cleanest way. ### Example handling SIGINT ```typescript import { Bus, BusInstance } from '@node-ts/bus-core' let bus: BusInstance const start = async () => { bus = Bus.configure().initialize() await bus.start() } /** * Listens for a signal interrupt and gracefully disposes the bus * before exiting. */ const listenForSigInt = () => { process.once('SIGINT', async () => { console.log('Received SIGINT, shutting down...') if (bus) { await bus.dispose() } }) } listenForSigInt() start().catch(console.err) ``` # Reference # Bus **Bus** is used as an entry point to configure a new [BusInstance](/reference/businstance). ### Bus.configure() Create a new [BusConfiguration](/reference/busconfiguration) in order to initialize a new [BusInstance](/reference/businstance). ```typescript import { Bus } from '@node-ts/bus-core' const busConfiguration = Bus.configure() ``` # BusConfiguration Creates a configuration in order to initialize a new [BusInstance](/reference/businstance). ## Methods ### withHandler(classHandler) Registers a class handler that receives a message and performs a unit of work. When **Bus** is initialized it will configure the transport to subscribe to the type of message handled by the handler and upon receipt will forward the message through to the `handle()` function. #### **Arguments**
ArgumentDescriptionDefault
classHandlerA class responsible for handling messages that implements HandlerNone
#### Example ```typescript import { Bus } from '@node-ts/bus-core' import { TestHandler } from './test-handler' Bus.configure().withHandler(TestHandler) ``` See also [events](/guide/messages/events), [commands](/guide/messages/commands). ### withHandler(functionHandler) Registers a function handler that receives a message and performs a unit of work. When **Bus** is initialized it will configure the transport to subscribe to the type of message handled by the function handler and upon receipt will forward the message to the function. #### Arguments
ArgumentDescriptionDefault
functionHandlerA functional handler mapping initialized using handlerForNone
#### Example ```typescript import { Bus, handlerFor } from '@node-ts/bus-core' import { TestEvent } from './test-event' Bus.configure().withHandler(handlerFor(TestEvent, event => {})) ``` See also [events](/guide/messages/events), [commands](/guide/messages/commands). ### withCustomHandler(messageHandler, customResolver) Registers a custom handler that receives messages from external systems, or messages that don't implement the **Message** interface from **@node-ts/bus-messages**. #### Arguments
ArgumentDescriptionDefault
messageHandlerA handler that receives the custom messageNone
customResolverA discriminator that determines if an incoming message should be mapped to this handlerNone
#### Example ```typescript import { Bus } from '@node-ts/bus-core' import { S3Event } from 'aws-sdk' Bus.configure() .withCustomHandler( async (event: S3Event) => console.log('Received S3 event', { event }), { resolveWith: event => event.Records && event.Records.length } ) ``` See also [System messages](/guide/messages/system-messages). ### withWorkflow(workflow) Registers a workflow definition so that all of the messages it depends on will be subscribed to and forwarded to the handlers inside the workflow. #### Arguments
ArgumentDescriptionDefault
workflowWorkflow definition to registerNone
#### Example ```typescript import { Bus } from '@node-ts/bus-core' import { TestWorkflow } from './test-workflow' Bus.configure().withWorkflow(TestWorkflow) ``` See also [Workflows](/guide/workflows). ### withTransport(transport) Configures **Bus** to use a different transport than the default **MemoryQueue.** #### Arguments
ArgumentDescriptionDefault
transportA configured transport to useNone
#### Example ```typescript import { Bus } from '@node-ts/bus-core' import { SqsTransport, SqsTransportConfiguration } from '@node-ts/bus-sqs' const sqsConfiguration: SqsTransportConfiguration = { // ... } const sqsTransport = new SqsTransport(sqsConfiguration) Bus.configure().withTransport(sqsTransport) ``` See also [Transports](/guide/transports). ### withLogger(loggerFactory) Configures **Bus** to use a different logging provider than the default consoler logger. #### Arguments
ArgumentDescriptionDefault
loggerFactoryA factory that creates a new loggerNone
#### Example ```typescript import { Bus } from '@node-ts/bus-core' import { CustomLogger } from './custom-logger' Bus.configure().withLogger((target: string) => new CustomLogger(target)) ``` See also [Loggers](/guide/loggers). ### withSerializer(serializer) 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. #### Arguments
ArgumentDescriptionDefault
serializerSerializer to useNone
#### Example ```typescript import { Bus } from '@node-ts/bus-core' import { ClassSerializer } from '@node-ts/bus-class-serializer' Bus.configure().withSerializer(new ClassSerializer()) ``` See also [Serializer](/guide/serializers). ### withPersistence(persistence) 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. #### Arguments
ArgumentDescriptionDefault
persistencePersistence provider to useNone
#### Example ```typescript import { Bus } from '@node-ts/bus-core' import { PostgresPersistence, PostgresConfiguration } from '@node-ts/bus-postgres' const postgresConfiguration: PostgresConfiguration = { connection: { connectionString: 'postgres://postgres:password@localhost:5432/postgres' }, schemaName: 'workflows' } const postgresPersistence = new PostgresPersistence(postgresConfiguration) Bus.configure().withPersistence(postgresPersistence) ``` See also [Persistence](/guide/persistence). ### withConcurrency(concurrency) Sets the message handling concurrency beyond the default value of 1, which will increase the number of messages handled in parallel.concurrency #### Arguments
ArgumentDescriptionDefault
concurrencyThe number of messages that can be handled in parallelNone
#### Example ```typescript import { Bus } from '@node-ts/bus-core' Bus.configure().withConcurrency(5) ``` ### withContainer(containerAdapter) ```typescript withContainer({ get (type: ClassConstructor) { return container.get(type) } }) ``` Use a local dependency injection/IoC container to resolve handlers and workflows. 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. #### Arguments
ArgumentDescriptionDefault
containerAdapterAn adapter that allows Bus to resolve class instances from the underlying IoC containerNone
#### Example ```typescript import { Bus } from '@node-ts/bus-core' import { Container } from 'inversify' const container = new Container() Bus.configure().withContainer({ get (type: ClassConstructor) { return container.get(type) } }) ``` See also [Dependency injection.](/guide/dependency-injection) ### withMessageReadMiddleware(middleware) ```typescript withMessageReadMiddleware ( messageReadMiddleware: Middleware> ) ``` Run custom middleware before/after the point a message is read from the transport and then dispatched to handlers and workflow handlers. #### Arguments
ArgumentDescriptionDefault
middlewareA middleware function that will be executed after a message is read from the transport and before it is dispatched to handlers.None
#### Example ```typescript import { Bus, Middleware, Next, TransportMessage } from '@node-ts/bus-core' const messageTimingMiddleware = async ( context: TransportMessage, next: Next ) => { const start = Date.now() await next() const end = Date.now() const durationMs = end - start console.log( 'Message handled', { messageName: context.domainMessage.$name, durationMs } ) } const bus = await Bus.configure() .withMessageReadMiddleware(messageTimingMiddleware) ``` See also [Middleware](/guide/middleware). ### withRetryStrategy(retryStrategy) ```typescript withRetryStrategy({ calculateRetryDelay (currentAttempt: number): number }) ``` Configure the bus to use a different retry strategy instead of the default. #### Arguments
ArgumentDescriptionDefault
retryStrategyAn implementation of RetryStrategy that calculates the delay between retrying failed messages.DefaultRetryStrategy
#### Example ```typescript const bus = await Bus.configure() .withRetryStrategy({ calculateRetryDelay (currentAttempt: number) { return Math.pow(2, currentAttempt) } }) .initialize() ``` See also [Retry Strategies](/guide/retry-strategies). ### initialize(\[options]) Initialize a configured **BusInstance**. This should be called after all options have been provided for the configuration. # BusInstance A configured instance of a bus used to publish messages. ## Properties ### beforeSend See [Lifecycle hooks](/guide/lifecycle-hooks). ### beforePublish See [Lifecycle hooks](/guide/lifecycle-hooks). ### onError See [Lifecycle hooks](/guide/lifecycle-hooks). ### afterReceive See [Lifecycle hooks](/guide/lifecycle-hooks). ### beforeDispatch See [Lifecycle hooks](/guide/lifecycle-hooks). ### afterDispatch See [Lifecycle hooks](/guide/lifecycle-hooks). ### state Gets the current state of the bus to see if it's started or stopped. ## Methods ### publish(event, \[messageAttributes]) Publishes an event to the bus, with an optional set of attributes to attach to the outgoing message. #### Arguments
ArgumentDescriptionDefault
eventThe message to publish to the underlying transportNone
messageAttributesAn optional set of attributes that will be sent with the outgoing eventNone
### send(command, \[messageAttributes]) Sends a command to the bus, with an optional set of attributes to attach to the outgoing message. #### Arguments
ArgumentDescriptionDefault
commandA message to send to the underlying transportNone
messageAttributesAn optional set of attributes that will be sent with the outgoingNone
### fail() Instructs the bus that the current message being handled cannot be processed even with retries and instead should immediately be routed to the dead letter queue. ### start() Instructs the bus to start reading messages from the underlying service queue and dispatching to message handlers. ### stop() Stops a bus that has been started by `.start()`. This will wait for all running workers to complete their current message handling contexts before returning. ### dispose() Stops and disposes all resources allocated to the bus, as well as removing all handler registrations. The bus instance cannot be used after this has been called. # Getting help For bugs and feature requests, create an issue at For general questions and discussion, join our discord at # Messages Messages are small pieces of data that get passed around between services. They can define an instruction to perform an action, or report that something has just occurred. Messages are sent by a publisher and received by 0 to many subscribers depending on the type of message and if your system requires it to be handled. # Events An event is a message emitted by the system when "something" happens. Again this could be a technical task being completed such as a **DatabaseBackedUp**, **LoadBalancerScaledOut** or as a result of changes in your business **CreditCardCharged**, **UserRegistered**, **PackageShipped**. {% hint style="info" %} Use natural language in the past-tense when naming an event since it represents a historic fact has taken place. This also helps improve the readability of your code, as the history of your application can be discussed in terms of the order of events. {% endhint %} ### Creating an Event Events are class definitions that extend from `Event`, eg: ```typescript import { Event } from '@node-ts/bus-messages' export class CreditCardCharged extends Event { /** * A unique name that identifies the message. This should be done in namespace style syntax, * ie: organisation/domain/event-name */ $name = 'my-app/accounts/credit-card-charged' /** * The contract version of this message. This can be incremented if this message changes the * number of properties etc to maintain backwards compatibility */ $version = 1 /** * A credit card was successfully charged * @param creditCardToken Identifies the card that was charged * @param amount The amount, in USD, that the card was charged for */ constructor ( readonly creditCardToken: string, readonly amount: number ) { } } ``` {% hint style="info" %} It's useful to declare all of your messages in a central package that can be shared amongst your publisher and subscriber services. {% endhint %} ### Publishing an Event Events can have 0-to-many different subscribers, who are generally interested in performing a next action as a result of the event being raised. Use `.publish()` to publish an event: ```typescript const creditCardCharged = new CreditCardCharged('abc', 123) // Publish a message. All subscribers will receive a copy await bus.publish(creditCardCharged) // Publish a message along with a set of attributes await bus.publish( creditCardCharged, { correlationId: 'tok-1adsfas-df1' } ) ``` ### Handling an Event Events get processed by a **Handler**. This is a function or a class function that receives the message as a parameter and performs an operation. When the handler returns the message is deleted from the queue. Implementing a function based handler ```typescript import { handlerFor } from '@node-ts/bus-core' // Function based handler const creditCardChargedHandler = handlerFor( CreditCardCharged, async (event: CreditCardCharged) => { // ... } ) ``` Implementing a class based handler ```typescript import { Handler } from '@node-ts/bus-core' // Class based handler class CreditCardChargedHandler implements Handler { messageType = CreditCardCharged async handle (event: CreditCardCharged) { // ... } } ``` Register the handler with the bus configuration ```typescript const bus = await Bus.configure() .withHandler(chargeCreditCardHandler) // Function based handler .withHandler(ChargeCreditCardHandler) // Class function based handler .initialize() ``` Remember to `.start()` the bus to start handling messages ```typescript await bus.start() ``` # Commands Commands are a type of message that represents an instruction to do work. These can be technical instructions such as **BackupDatabase**, **ScaleOutLoadBalancer** or modelled after your business domains like **PlaceOrder**, **ShipPackage**. {% hint style="info" %} Use plain english when naming commands. This helps understanding what a command will do once it's processed. {% endhint %} To implement a command, extend the `Command` class and add any relevant fields ```typescript import { Command } from '@node-ts/bus-messages' export class ChargeCreditCard extends Command { /** * A unique name that identifies the message. This should be done in namespace style syntax, * ie: organisation/domain/command-name */ $name = 'my-app/accounts/charge-credit-card' /** * The contract version of this message. This can be incremented if this message changes the * number of properties etc to maintain backwards compatibility */ $version = 1 /** * Create a charge on a credit card * @param creditCardToken Identfies the card to charge * @param amountThe amount, in USD, to charge the card */ constructor ( readonly creditCardToken: string, readonly amount: number ) { } } ``` A commands are sent to a single service for processing, and generally result in the publication of one or more [events](/guide/messages/events). {% hint style="info" %} It's useful to declare all of your messages in a central package that can be shared amongst your publisher and subscriber services. {% endhint %} ### Sending a command Commands should only be processed by a single service, unlike events which may have multiple subscribers. Command processors usually receive a command, process it, and emit an event as a result of the operation completing. Use `.send()` to send a command: ```typescript const chargeCreditCard = new ChargeCreditCard('abc', 123) // Send a command. This will be handled by a single subscriber await bus.send(chargeCreditCard) // Send a message along with a set of attributes await bus.send( chargeCreditCard, { correlationId: 'tok-1adsfas-df1' } ) ``` ### Handling a Command Comments get processed by a **Handler**. This is a function or a class function that receives the message as a parameter and performs an operation. When the handler returns the message is deleted from the queue. Implementing a function based handler ```typescript import { handlerFor } from '@node-ts/bus-core' // Function based handler const chargeCreditCardHandler = handlerFor( ChargeCreditCard, async (event: ChargeCreditCard) => { // ... } ) ``` Implementing a class based handler ```typescript import { Handler } from '@node-ts/bus-core' // Class based handler class ChargeCreditCardHandler implements Handler { messageType = ChargeCreditCard async handle (event: ChargeCreditCard) { // ... } } ``` Register the handler with the bus configuration ```typescript const bus = await Bus.configure() .withHandler(chargeCreditCardHandler) // Function based handler .withHandler(ChargeCreditCardHandler) // Class function based handler .initialize() ``` Remember to `.start()` the bus to start handling messages ```typescript await bus.start() ``` # System messages System messages are message that originate externally to your app. These can be things like an event from S3 indicating an object has been created, an email receipt on a support mailbox, or a disk space threshold has been reached on an important server. Because system messages are created and sent by external systems, it's not possible to enforce **@node-ts/bus** conventions on the structure of the message. Instead messages are consumed as is, with custom mapping code used to identify a system message and route it to its handler. {% hint style="info" %} Having an adapter handler that receives a system message and then maps it to a locally declared message that's then published helps ensure your app doesn't get corrupted with language and conventions of external systems. {% endhint %} ### Creating a System Message System messages are class definitions. If the system publishing the message doesn't expose a class definition for the message they're publishing then it can be created. ```typescript /** * A message that is published by S3 each time an action has occurred on an object */ export class S3Event { readonly Records: { eventSource: 'aws:s3', eventName: string, s3: { object: { key: string } } }[] } ``` ### Handling System Messages System messages must be subscribed to manually. This usually means configuring the system to send to a topic that the application is subscribed to. This will ensure that messages sent by the external system will be routed to and read from the service queue. Handler registration is done using the `.withCustomHandler()` function on configuration. This accepts the handler, a custom `resolveWith` property that identifies incoming messages as being the type the handler should receive, and a `topicIdentifier` that **@node-ts/bus** will subscribe the application service queue to. ```typescript import { S3Event } from 'contracts' await Bus.configure() .withCustomHandler( async (event: S3Event) => console.log('Received S3 event', { event }), { resolveWith: event => event.Records && event.Records.length && event.Records[0].eventSource === 'aws:s3', topicIdentifier: 'arn:aws:sns:us-east-1:000000000000:s3-object-created' } ).initialize() ``` Remember to `.start()` the bus to start handling messages ```typescript await bus.start() ``` # Message attributes Additional metadata can be added to any type of message as attributes alongside the actual message. When sending via a transport (eg: SQS, RabbitMQ) the message is sent in a message envelope. [Commands](/guide/messages/commands) and [events](/guide/messages/events) are serialized into the message body, whilst attributes are added to the message header. Attributes are designed to hold data that is related to technical concerns of routing the message, or auditing/logging information such as details around the originator. # Correlation id An identifier that can be used to relate or correlate messages together. This value is sticky meaning that when a message is received with a correlation id, any messages that are sent as a result will contain the same correlation id. Correlation ids are a useful mechanism for logging and tracking message flows through the system. They're also the default mechanism to correlate messages orchestrated by [workflows](/guide/workflows) to continue the next step in a process. ```typescript const start = () => { const bus = await Bus.configure() .withHandler( ChargeCreditCard, async () => await bus.publish(new CreditCardCharged()) ) .initialize() await bus.start() await bus.send( new ChargeCreditCard(), { correlationId: 'cd091b26-f0e6-43fb-9962-c06786948e26' } ) } ``` In this example, a **ChargeCreditCard** command is being sent with a correlation id ('cd091b26-f0e6-43fb-9962-c06786948e26'). This will be handled and an event called **CreditCardCharged** published as a result. This event will have a correlation id of the same value. # Attributes Arbitrary attributes can be sent with each message. Attributes support `string | number | boolean` values. These can be provided by using the `attributes` property when sending a message. ```typescript await bus.send( new ChargeCreditCard(), { attributes: { ip: '229.40.202.156', attempt: 0, automatic: true } } ) ``` Attributes can be accessed as the second parameter of the handler. ```typescript import { MessageAttributes } from '@node-ts/bus-messages' await Bus.configure() .withHandler( ChargeCreditCard, async (_: ChargeCreditCard, { attributes }: MessageAttributes) => console.log(attributes) ) .initialize() ``` # Sticky attributes Sticky attributes are similar to regular attributes except they will be copied to any message immediately or subsequently sent as a result of the original message being processed. These can be sent by specifying a value for `stickyAttributes` ```typescript await bus.send( new ChargeCreditCard(), { stickyAttributes: { ip: '229.40.202.156', attempt: 0, automatic: true } } ) ``` Sticky attributes can be accessed using the `stickyAttribtues` of the second parameter of a handler. ```typescript import { MessageAttributes } from '@node-ts/bus-messages' await Bus.configure() .withHandler( ChargeCreditCard, async (_: ChargeCreditCard, { stickyAttributes }: MessageAttributes) => console.log(attributes) ) .initialize() ``` # Workflows Workflows help orchestrate processes in a distributed environment. Oftentimes business processes are a series of steps to be taken. Each step is a command that's sent, with the next step being run after the prior one completes. This can be difficult in a distributed environment as you can't choose which node will receive events to trigger then next command, and so you can't keep the state of the workflow locally. In **@node-ts/bus**, the state of a workflow is stored in a [persistence](/guide/persistence) technology like Postgres. Because the state is stored locally it means any node can consume the state and decide which step should happen next. # Creating a workflow A **workflow** is defined as a class that contains a number of **steps** using functions, and the **workflow state** modelled as a separate class. A workflow can be started by one or more different types of messages. Each time a workflow is started, a new workflow state is created. This state will be available to each of the steps in the workflow, and each step can mutate the state by returning some of its properties. ### Creating Create a workflow state by declaring a class that extends **WorkflowState**. The `$name` property should be unique among all of your workflows. ```typescript // fulfilment-workflow-state.ts import { WorkflowState } from '@node-ts/bus-core' export class FulfilmentWorkflowState extends WorkflowState { static NAME = 'FulfilmentWorkflowState' $name = FulfilmentWorkflowState.NAME } ``` Create the workflow definition that extends **Workflow**, passing it a generic parameter of the workflow state. By extending this class you'll need to provide a`configureWorkflow(mapper: WorkflowMapper): void` implementation. This mapper is used to configure how messages are dispatched to functions in the workflow. ```typescript // example-workflow.ts import { Workflow } from '@node-ts/bus-core' export class FulfilmentWorkflow extends Workflow { configureWorkflow ( mapper: WorkflowMapper ): void { mapper.withState(FulfilmentWorkflowState) } } ``` Lastly register the workflow with the bus configuration on startup. ```typescript const bus = await Bus.configure() .withWorkflow(FulfilmentWorkflow) .initialize() ``` # Starting Instances of a workflow are started by one or more messages. When one of these types of messages are received a new workflow state is created and execution of the workflow begins. Declare a handler for the message that starts your workflow, and then map it using the **WorkflowMapper**. ```typescript import { Workflow } from '@node-ts/bus-core' export class FulfilmentWorkflow extends Workflow { configureWorkflow ( mapper: WorkflowMapper ): void { mapper .withState(FulfilmentWorkflowState) // Start a new workflow when an `ItemPurchased` event is received .startedBy(ItemPurchased, 'shipItem') } // Handles an `ItemPurchased` event async shipItem (event: ItemPurchased) { // ... } } ``` In this case, when a **ItemPurchased** event is received, it will start a new workflow and dispatch the message to the **shipItem** handler. # Handling After a workflow has [started](/guide/workflows/starting) it will often wait until a new message is read from the queue so that it can perform the next step in its process. This is done by configuring a handler for that message, and defining a discriminator that will determine which workflow instance should be activated for that message. Handlers are created by declaring a function on the workflow and then providing a `.when()` mapping on the **WorkflowMapper**. ```typescript import { Workflow } from '@node-ts/bus-core' export class FulfilmentWorkflow extends Workflow { configureWorkflow ( mapper: WorkflowMapper ): void { mapper .withState(FulfilmentWorkflowState) // ... // When the item is shipped, email the customer their receipt .when(ItemShipped, 'emailReceipt') } // Handles an `ItemShipped` event async emailReceipt (event: ItemShipped) { // ... } } ``` ### Default mapping When a workflow is started it is assigned a **$workflowId** that is persisted against the state and cannot be changed. This **$workflowId** is attached to all messages sent from within the workflow in [sticky attributes](/guide/message-attributes/sticky-attributes), which propagates as subsequent messages are sent. Handling these messages from within a workflow will locate the workflow state based on the value of **$workflowId**. This example uses the default mapping in the `.when()` handler ```typescript import { Workflow, BusInstance } from '@node-ts/bus-core' export class FulfilmentWorkflow extends Workflow { constructor (bus: BusInstance): {} configureWorkflow ( mapper: WorkflowMapper ): void { mapper .withState(FulfilmentWorkflowState) .startedBy(ItemPurchased, 'shipItem') // When the item is shipped, email the customer their receipt .when(ItemShipped, 'emailReceipt') } async shipItem (event: ItemPurchased): { await this.bus.send(new ShipItem(event.itemId)) } // Handles an `ItemShipped` event async emailReceipt (event: ItemShipped) { // ... } } ``` What happens is: 1. A new workflow is started on receipt of an **ItemPurchased** event, with a new **$workflowId** value on the state 2. The **shipItem** handler will send a new **ShipItem**. The workflow will automatically attach the **$workflowId** into the sticky attributes of the outgoing message 3. The command handler for **ShipItem** will process the request and publish an **ItemShipped** event. Because **$workflowId** is present on the sticky attributes of the incoming command, it will also be attached to the sticky attributes of the outgoing event 4. The **ItemShipped** event is received and since this workflow uses a default handler for the message, the value of **$workflowId** on the incoming sticky attributes will be used to lookup the workflow state instance 5. The **ItemShipped** event is then routed to the **emailReceipt** handler for the workflow instance Default mappings are a simple way to map workflow handlers when the next step is based on the outcome of a command sent by the workflow. ### Mapping via message properties Messages can be mapped to workflow handlers by matching the value of a property of a message to a property on the workflow state. This is done by providing a `lookup` and `mapsTo` configuration for the message on the **WorkflowMapper** ```typescript import { Workflow, BusInstance, WorkflowState } from '@node-ts/bus-core' type uuid = string class FulfilmentWorkflowState extends WorkflowState { $name = 'FulfilmentWorkflowState' itemId: uuid } export class FulfilmentWorkflow extends Workflow { constructor (bus: BusInstance): {} configureWorkflow ( mapper: WorkflowMapper ): void { mapper .withState(FulfilmentWorkflowState) .startedBy(ItemPurchased, 'shipItem') // When the item is shipped, email the customer their receipt .when( ItemShipped, 'emailReceipt', { // When an `ItemShipped` event is received, get the value of `itemId`... lookup: itemShippedEvent => itemShippedEvent.itemId, // ...and grab the workflow with a matching 'itemId' value on the FulfilmentWorkflowState mapsTo: 'itemId' } ) } async shipItem (event: ItemPurchased): { await this.bus.send(new ShipItem(event.itemId)) // Persist the updated `itemId` value on the workflow sstate return { itemId: event.itemId } } // Handles an `ItemShipped` event async emailReceipt (event: ItemShipped) { // ... } } ``` In this example, the workflow is started when an `ItemPurchased` event is received. The workflow sends a command to `ShipItem` and persists the `itemId` in the workflow state. Eventually when an `ItemShipped` event is received, the `lookup` function grabs the value of `itemId` and tells the bus to get the workflow that has a matching value for `itemId` in the workflow state by using `mapsTo`. Mapping via message properties is useful when the message being handled is a result of an action not triggered by the workflow. ### Mapping via message attributes Messages can also be mapped to workflow handlers using values held in the message attributes matched to values in the workflow state. This is done by providing values for `lookup` and `mapsTo` in the handler mapping ```typescript import { Workflow, BusInstance } from '@node-ts/bus-core' export class FulfilmentWorkflow extends Workflow { constructor (bus: BusInstance): {} configureWorkflow ( mapper: WorkflowMapper ): void { mapper .withState(FulfilmentWorkflowState) .startedBy(ItemPurchased, 'shipItem') // When the item is shipped, email the customer their receipt .when( ItemShipped, 'emailReceipt', { // When an `ItemShipped` event is received, get the itemId from the message attributes... lookup: (_, { attributes }) => attributes.itemId, // ...and grab the workflow with a matching 'itemId' value on the workflow state mapsTo: 'itemId' } ) } async shipItem ({ itemId }: ItemPurchased): { await this.bus.send( new ShipItem(itemId), { attributes: { itemId } } ) // Persist the updated `itemId` value on the workflow sstate return { itemId } } // Handles an `ItemShipped` event async emailReceipt (event: ItemShipped) { // ... } } ``` # State The workflow state keeps track of the state of the workflow as it progresses. The state is stored in the configured [persistence](/guide/persistence), and can be used to map incoming messages to [handlers](/guide/workflows/handling). ### Defining the state A workflow state is created by extending **WorkflowState.** The `$name` property should be unique among all of your workflows. ```typescript import { WorkflowState } from '@node-ts/bus-core' export class FulfilmentWorkflowState extends WorkflowState { static NAME = 'FulfilmentWorkflowState' $name = FulfilmentWorkflowState.NAME // Set of user-defined state properties itemId: string customerId: string status: 'posting-item' | 'sending-receipt' | 'complete' } ``` ### Accessing the state The state is available as the second parameter to all handlers of a workflow. For example: ```typescript import { Workflow } from '@node-ts/bus-core' export class FulfilmentWorkflow extends Workflow { configureWorkflow ( mapper: WorkflowMapper ): void { mapper .withState(FulfilmentWorkflowState) // ... .when(ItemShipped, 'emailReceipt') } // Workflow state is passed in as the second parameter to a workflow handler async emailReceipt (_: ItemShipped, state: FulfilmentWorkflowState) { // ... } } ``` ### Updating the state The state cannot be modified directly within a handling scope but can be updated by returning the intended changes from a handler. Returning an updated state is optional, and if omitted then no changes to the state will be persisted. ```typescript import { Workflow } from '@node-ts/bus-core' export class FulfilmentWorkflow extends Workflow { configureWorkflow ( mapper: WorkflowMapper ): void { mapper .withState(FulfilmentWorkflowState) // ... .when(ItemShipped, 'emailReceipt') } // Workflow state is passed in as the second parameter to a workflow handler async emailReceipt (_: ItemShipped, state: FulfilmentWorkflowState) { return { status: 'sending-receipt' } } } ``` ### Discarding state There are times when the workflow data shouldn't persist after a message has been handled. This is particularly relevant in cases where a workflow should only handle a message under certain circumstances. For example, if your workflow is started by an **S3ObjectCreated** event, but should only create a new workflow if the object key is prefixed with `/documents`, then this can be achieved by returning `this.discardWorkflow()` in the workflow like so: ```typescript import { Workflow, BusInstance } from '@node-ts/bus-core' export class ProcessDocumentWorkflow extends Workflow { constructor (bus: BusInstance) {} configureWorkflow ( mapper: WorkflowMapper ): void { mapper .withState(ProcessDocumentWorkflowState) .startedBy(S3ObjectCreated, 'readDocument') } async readDocument (event: S3ObjectCreated) { if (event.s3Key.indexOf('/documents') === 0 { await this.bus.send(new ReadDocument(event.s3Key)) } else { // Ignore this message and avoid persisting the workflow state return this.discardWorkflow() } } } ``` # Completing A workflow is completed by returning `this.completeWorkflow()` from within a handler. This will set the state of the workflow to `completed` and will no longer be activated by any incoming messages. ```typescript import { Workflow } from '@node-ts/bus-core' export class FulfilmentWorkflow extends Workflow { configureWorkflow ( mapper: WorkflowMapper ): void { mapper .withState(FulfilmentWorkflowState) // ... .when(ReceiptSent, 'complete') } async complete (_: ReceiptSent) { return this.completeWorkflow() } } ``` Final changes to the workflow state can be passed in as a parameter to `completeWorkflow()` if desired ```typescript export class FulfilmentWorkflow extends Workflow { configureWorkflow ( mapper: WorkflowMapper ): void { mapper .withState(FulfilmentWorkflowState) // ... .when(ReceiptSent, 'complete') } async complete (_: ReceiptSent) { return this.completeWorkflow({ status: 'complete' }) } } ``` # Example This example models an example fulfilment process once a customer has purchased an item from an online store. ```typescript // fulfiment-workflow-state.ts import { WorkflowState } from '@node-ts/bus-core' export class FulfilmentWorkflowState extends WorkflowState { static NAME = 'FulfilmentWorkflowState' $name = FulfilmentWorkflowState.NAME itemId: string customerId: string status: 'email-receipt' | 'shipping-item' | 'complete' } ``` ```typescript import { BusInstance } from '@node-ts/bus-core' export class FulfilmentWorkflow extends Workflow { constructor (bus: BusInstance) {} configureWorkflow ( mapper: WorkflowMapper ): void { mapper .withState(FulfilmentWorkflowState) .startedBy(ItemPurchased, 'shipItem') .when(ItemShipped, 'emailReceipt') .when(ReceiptEmailed, 'complete') } async shipItem ({ itemId, customerId }: ItemPurchased) { await this.bus.send(new ShipItem(itemId, customerId)) return { itemId, customerId, status: 'shipping-item' } } async emailReceipt (_: ItemShipped, { itemId, customerId }: FulfilmentWorkflowState) { await this.bus.send(new EmailReceipt(itemId, customerId)) return { status: 'emailing-receipt' } } async complete () { return this.completeWorkflow({ status: 'complete' }) } } ``` # Transports A transport is a message queue that's able to receive messages. By default **@node-ts/bus** runs with an in memory queue transport, which is only really useful for development environments. When running in production, choose from one of the following officially supported transports: * [Amazon Simple Queue Service (SQS)](/guide/transports/amazon-sqs) * [Particular RabbitMQ](/guide/transports/rabbitmq) * [Redis](/guide/transports/redis) These transports are interchangeable and you write your message and handler definitions agnostic of the underlying transport. # RabbitMQ RabbitMQ is an AMQP compatible transport that's officially supported by **@node-ts/bus**. Once configured, **@node-ts/bus** will create all necessary queues and exchanges to support the handlers of the application. ### Installation Install the **@node-ts/bus-rabbitmq** npm package ```bash npm i @node-ts/bus-rabbitmq ``` Once installed, configure a new **RabbitMqTransport** and provide it to the bus configuration ```typescript import { Bus } from '@node-ts/bus-core' import { RabbitMqTransport, RabbitMqTransportConfiguration } from '@node-ts/bus-rabbitmq' const rabbitConfiguration: RabbitMqTransportConfiguration = { queueName: 'accounts-application-queue', connectionString: 'amqp://guest:guest@localhost', maxRetries: 5 } const rabbitMqTransport = new RabbitMqTransport(rabbitConfiguration) await Bus .configure() .withTransport(rabbitMqTransport) .initialize() ``` # Amazon SQS SQS is a fully managed queue provider from AWS that's officially supported by **@node-ts/bus**. Once configured, **@node-ts/bus** will automatically create all SNS topics, SQS queues and manage subscriptions for the handlers of your application. ### Installation Install the **@node-ts/bus-sqs** package ```bash npm install @node-ts/bus-sqs ``` Configure the transport and provide this to the bus configuration ```typescript import { Bus } from '@node-ts/bus-core' import { SqsTransport, SqsTransportConfiguration } from '@node-ts/bus-sqs' const sqsConfiguration: SqsTransportConfiguration = { awsRegion: AWS_REGION, awsAccountId: AWS_ACCOUNT_ID, queueName: 'service-queue', deadLetterQueueName: `service-queue-dead-letter` } const sqsTransport = new SqsTransport(sqsConfiguration) await Bus .configure() .withTransport(sqsTransport) .initialize() ``` # Redis Redis can be used as a message transport by installing the **@node-ts/bus-redis** package. This will configure Redis to use lists as de-facto message queues and route messages according to your application handlers. ### Installation Install the **@node-ts/bus-redis** package ```bash npm install @node-ts/bus-redis ``` Configure the Redis transport and provide it to the bus configuration ```typescript import { Bus } from '@node-ts/bus-core' import { RedisTransport, RedisTransportConfiguration } from '@node-ts/bus-redis' const redisTransportConfiguration: RedisTransportConfiguration = { queueName: 'accounts-application-queue', connectionString: 'redis://127.0.0.1:6379', maxRetries: 3 } const redisTransport = new RedisTransport(redisTransportConfiguration) await Bus .configure() .withTransport(redisTransport) .initialize() ``` # Custom transports If you want to use a message queue that's not officially supported by the transport list, you can provide an adapter layer by implementing the **Transport** interface from **@node-ts/bus-core**. The following interface definition lists the functionality that must be implemented in order for the transport to be compatible. ```typescript /** * A transport adapter interface that enables the service bus to use a messaging technology. */ export interface Transport { /** * Publishes an event to the underlying transport. This is generally done to a topic or some other * mechanism that consumers can subscribe themselves to * @param event A domain event to be published * @param messageOptions Options that control the behaviour around how the message is sent and * additional information that travels with it. */ publish (event: TEvent, messageOptions?: MessageAttributes): Promise /** * Sends a command to the underlying transport. This is generally done to a topic or some other * mechanism that consumers can subscribe themselves to * @param command A domain command to be sent * @param messageOptions Options that control the behaviour around how the message is sent and * additional information that travels with it. */ send (command: TCommand, messageOptions?: MessageAttributes): Promise /** * Forwards @param transportMessage to the dead letter queue. The message must have been read in from the * queue and have a receipt handle. */ fail (transportMessage: TransportMessage): Promise /** * Forwards @param transportMessage to the dead letter queue. The message must have been read in from the * queue and have a receipt handle. */ fail (transportMessage: TransportMessage): Promise /** * Fetch the next message from the underlying queue. If there are no messages, then `undefined` * should be returned. * * @returns The message construct from the underlying transport, that inclues both the raw message envelope * plus the contents or body that contains the `@node-ts/bus-messages` message. */ readNextMessage (): Promise | undefined> /** * Removes a message from the underlying transport. This will be called once a message has been * successfully handled by any of the message handling functions. * @param message The message to be removed from the transport */ deleteMessage (message: TransportMessage): Promise /** * Returns a message to the queue for retry. This will be called if an error was thrown when * trying to process a message. * @param message The message to be returned to the queue for reprocessing */ returnMessage (message: TransportMessage): Promise /** * An optional function that is called before startup that will provide core dependencies * to the transport. This can be used to fetch loggers, registries etc that are used * in initialization steps * @param coreDependencies */ prepare (coreDependencies: CoreDependencies): void /** * An optional function that will be called on startup. This gives a chance for the transport * to establish any connections to the underlying infrastructure. */ connect? (): Promise /** * An optional function that will be called on shutdown. This gives a chance for the transport * to close any connections to the underlying infrastructure. */ disconnect? (): Promise /** * An optional function that will be called when the service bus is starting. This is an * opportunity for the transport to see what messages need to be handled so that subscriptions * to the topics can be created. * @param handlerRegistry The list of messages being handled by the bus that the transport needs to subscribe to. */ initialize? (handlerRegistry: HandlerRegistry): Promise /** * An optional function that will be called when the service bus is shutting down. This is an * opportunity for the transport to close out any open requests to fetch messages etc. */ dispose? (): Promise } ``` Once your transport has implemented this interface, it can be provided to the bus on configuration ```typescript import { Bus } from '@node-ts/bus-core' import { MyTransport } from './my-transport' const myTransport = new MyTransport() await Bus .configure() .withTransport(myTransport) .intialize() ``` # Persistence By default **@node-ts/bus-core** uses an in-memory persistence to store the state of workflows. This is intended for development only and a durable persistence should be used in all other situations. If you are not using workflows in your service, then you do not need to configure a persistence. # Postgres **@node-ts/bus-postgres** provides a persistence adapter based on Postgres that's used when storing workflow state. ### Installation Install the **@node-ts/bus-postgres** package ``` npm install @node-ts/bus-postgres ``` Create a new **PostgresPersistence** instance and provide it to the bus configuration. ```typescript import { Bus } from '@node-ts/bus-core' import { PostgresPersistence, PostgresConfiguration } from '@node-ts/bus-postgres' const postgresConfiguration: PostgresConfiguration = { connection: { connectionString: 'postgres://postgres:password@localhost:5432/postgres' }, schemaName: 'workflows' } const postgresPersistence = new PostgresPersistence(postgresConfiguration) await Bus .configure() .withPersistence(postgresPersistence) .initialize() ``` # MongoDB **@node-ts/bus-mongodb** provides a persistence adapter based on Mongo DB that's used when storing workflow state. ### Installation Install the **@node-ts/bus-mongodb** package ``` npm install @node-ts/bus-mongodb ``` Create a new **MongodbPersistence** instance and provide it to the bus configuration. ```typescript import { Bus } from '@node-ts/bus-core' import { MongodbPersistence, MongodbConfiguration } from '@node-ts/bus-mongodb' const configuration: MongodbConfiguration = { connection: 'mongodb://localhost:27017', databaseName: 'workflows' } const mongodbPersistence = new MongodbPersistence(configuration) // Configure bus to use mongodb as a persistence const run = async () => { await Bus .configure() .withPersistence(mongodbPersistence) .initialize() } run.then(() => void) ``` # Creating a persistence You can use your own persistence technology by implementing the **Persistence** interface and providing it as a configuration to bus. The **Persistence** interface has the following properties that need to be implemented ```typescript /** * Infrastructure that provides the ability to persist workflow state for long running processes */ export interface Persistence { /** * An optional function that is called before startup that will provide core dependencies * to the persistence. This can be used to fetch loggers etc that are used * in initialization steps * @param coreDependencies */ prepare (coreDependencies: CoreDependencies): void /** * If provided, initializes the persistence implementation. This is where database connections are * started. */ initialize? (): Promise /** * If provided, will dispose any resources related to the persistence. This is where things like * closing database connections should occur. */ dispose? (): Promise /** * Allows the persistence implementation to set up its internal structure to support the workflow state * that it will be persisting. Typically for a database this could mean setting up the internal table * schema to support persisting of each of the workflow state models. */ initializeWorkflow ( workflowStateConstructor: ClassConstructor, messageWorkflowMappings: MessageWorkflowMapping[] ): Promise /** * Retrieves all workflow state models that match the given `messageMap` criteria * @param workflowStateConstructor The workflow model type to retrieve * @param messageMap How the message is mapped to workflow state models * @param message The message to map to workflow state * @param includeCompleted If completed workflow state items should also be returned. False by default */ getWorkflowState ( workflowStateConstructor: ClassConstructor, messageMap: MessageWorkflowMapping, message: MessageType, messageOptions: MessageAttributes, includeCompleted?: boolean ): Promise /** * Saves a new workflow state model or updates an existing one. Persistence implementations should take care * to observe the change in `$version` of the workflow state model when persisting to ensure race conditions * don't occur. */ saveWorkflowState ( workflowState: WorkflowStateType ): Promise } ``` Once this is implemented, provide it to the bus configuration ```typescript import { Bus } from '@node-ts/bus-core' import { MyPersistence } from './my-persistence' const myPersistence = new MyPersistence() await Bus .configure() .withPersistsence(myPersistence) .initialize() ``` # Serializers **@node-ts/bus-core** by default uses a naive serializer that leverages the built-in `JSON.parse()` and `JSON.stringify()` functions. This is used to convert messages to a serialized form when publishing them to the transport, and then deserializing them when they're read from the transport. While this is fine for the simplest of cases, it suffers from the normal problems of deserializing into strong types. For example consider the following message ```typescript import { Command } from '@node-ts/bus-messages' export class RegisterUser extends Command { constructor ( readonly dateOfBirth: Date ) {} } ``` If this is run through the default serializer the effect will be the same as: ```typescript const command = new RegisterUser(new Date()) const plainCommand = JSON.parse(JSON.stringify(command)) plainCommand.getDate() // ERROR - getDate() does not exist on plainCommand ``` The default serializer is not recommended for serious projects given these limitations, and [class serializer](/guide/serializers/class-serializer) should be used where possible. # Class serializer **@node-ts/bus-class-serializer** leverages [class transformer](https://www.npmjs.com/package/class-transformer) to provide serialization/deserialization to class instances. This means that class based properties of messages like javascript Dates can be consumed directly without having to do additional custom deserialization in each handler. ### Installation Install **@node-ts/bus-class-serializer** and its dependencies ```bash npm install @node-ts/bus-class-serializer reflect-metadata class-transformer ``` Configure bus to use the serializer ```typescript import { Bus } from '@node-ts/bus-core' import { ClassSerializer } from '@node-ts/bus-class-serializer' Bus .configure() .withSerializer(new ClassSerializer()) .initialize() ``` {% hint style="info" %} This package relies on [class transformer](https://www.npmjs.com/package/class-transformer) that requires [reflect-metadata](https://www.npmjs.com/package/reflect-metadata) to be installed and called at the start of your application before any other imports. Please follow their guides on how to configure your app and contracts to serialize correctly. {% endhint %} ### Strongly typed messages Once the **@node-ts/bus-class-serializer** has been installed and configured, you can consume class transformer notation for object properties of your messages. For example ```typescript import { Command } from '@node-ts/bus-messages' import { Type } from 'class-transformer' class Update extends Command { // Provide a @Type hint so that class-transformer can deserialize this at runtime @Type(() => Date) readonly date: Date constructor ( date: Date ) { this.date = date } } ``` # Loggers **@node-ts/bus-core** by default uses the [debug](https://www.npmjs.com/package/debug) library in order to log details to STDOUT. All log messages that originate from this library are named with a `@node-ts/` prefix. In order to get full debug output, set the `DEBUG` environment variable to `@node-ts/*` For example: ```shell DEBUG="@node-ts/*" npm run dev ``` If you want to use your own logging provider and change the output format etc, this can be changed with any other logger by providing an adapter layer for a [custom logger](/guide/loggers/custom-loggers). # Custom loggers To use your own logger with **@node-ts/bus**, create an adapter class by implementing the **Logger** interface. In this example [winston](https://www.npmjs.com/package/winston) is being used ```typescript import { Logger } from '@node-ts/bus-core' import winston from 'winston' export class WinstonLogger implements Logger { private winstonLogger: winston.Logger constructor ( name: string // This is the name of the class the logger will be injected into ) { this.winstonLogger = winston.createLogger({ format: winston.format.json(), defaultMeta: { name } }) } debug (message: string, meta?: object): void { this.winstonLogger.debug(message, meta) } trace (message: string, meta?: object): void { this.winstonLogger.verbose(message, meta) } info (message: string, meta?: object): void { this.winstonLogger.info(message, meta) } warn (message: string, meta?: object): void { this.winstonLogger.warn(message, meta) } error (message: string, meta?: object): void { this.winstonLogger.error(message, meta) } fatal (message: string, meta?: object): void { this.winstonLogger.crit(message, meta) } } ``` Configure **@node-ts/bus-core** to use your log adapter ```typescript import { Bus } from '@node-ts/bus-core' import { WinstonLogger } from './winston-logger' ​ async function run () { const bus = await Bus.configure() .withLogger((target: string) => new WinstonLogger(target)) .initialize() } ``` # Middleware MIddleware can be added between when the message is read from a transport and before it is dispatched to handlers. ### Uses in telemetry Telemetry providers like AWS Xray, New Relic, Data Dog, etc can all be used to profile and report message processing times by using middleware. This is useful to troubleshoot low message processing rates and identify message types that are taking the longest to process and would benefit from performance tuning. The following shows how to integrate AWS Xray to profile how messages are handled. ```typescript import AWSXRay from 'aws-xray-sdk' import { Bus } from '@node-ts/bus-core' import { Message } from '@node-ts/bus-messages' const bus = await Bus.configure() .withMessageReadMiddleware(async (context, next) => { const messageName = (context.domainMessage as Message).$name const segment = new AWSXRay.Segment('my-service') const subSegment = segment.addNewSubSegment(messageName) await next() subSegment.close() segment.close() }) .initialize() ``` ### Uses in logging context Middleware is also useful when adding context of the message handling scope to each log produced in the handling cycle. An example of this is appending a message's `correlationId` to the metadata of each log. This can be done by using [async\_hooks](https://nodejs.org/docs/latest-v16.x/api/async_hooks.html) that attach data to the promise scope that can be accessed by any logging request inside of it. ```typescript import { Bus } from '@node-ts/bus-core' import * as asyncHooks from 'async_hooks' type CorrelationId = string // Requests to log() would look up the executionId in this map and attach the correlationId export const handlingContext = new Map() const bus = await Bus.configure() .withMessageReadMiddleware(async (context, next) => { const correlationId = context.attributes.correlationId const executionId = asyncHooks.executionAsyncId() handlingContext.set(executionId, correlationId) await next() handlingContext.delete(executionId) }) .initialize() ``` # Lifecycle hooks **@node-ts/bus** exposes a number of lifecycle hooks that can be subscribed to. These are **EventEmitter** instances that follow the standard node `.on()` and `.off()` notation. ### Hooks **beforeSend** called just before a command is sent to the underlying transport ``` bus.beforeSend.on(({ command, attributes }) => {}) ``` **beforePublish** called just before a event is published to the underlying transport ``` bus.beforePublish.on(({ event, attributes }) => {}) ``` **afterReceive** called after a message has been read from the queue, and before it is dispatched to handlers ``` bus.afterReceive.on(transportMessage => {}) ``` **beforeDispatch** called before a message is dispatched to its handlers ``` bus.beforeDispatch.on(({ message, attributes, handlers }) => {}) ``` **afterDispatch** called after a message has been successfully handled and the message deleted from the transport ``` bus.afterDispatch.on(({ message, attributes }) => {}) ``` **onError** called when an error occurred reading/dispatching/handling a message ``` bus.onError.on(({ message, error, attributes, rawMessage }) => {}) ``` # Retry Strategies Message retry strategies allow you to specify how much of a delay should occur before retrying a message. Delays between retries can be useful when messages fail handling due to race conditions, service unavailability, or concurrency reasons. By default, @node-ts/bus uses a `DefaultRetryStrategy` that exponentially increases the delay between each retry attempt. Additionally, it will introduce a random variance of 10% for each delay to help unblock messages that are failing when processed at the same time. Additional strategies can be implemented to suit your application. ## Choosing a Retry Strategy A retry strategy can be provided to the bus configuration on initialization by using `.withRetryStrategy()` For example: ```typescript const bus = await Bus.configure() .withRetryStrategy(DefaultRetryStrategy) .initialize() ``` ## Custom retry strategies A custom retry strategy can be provided by implementing the `RetryStrategy` from `` @node-ts/bus-core.` `` {% code title="retry-strategy.ts" %} ```typescript export type Milliseconds = number /** * Defines how a message retry strategy is to be implemented that calculates the delay between subsequent * retries of a message. */ export interface RetryStrategy { /** * Calculate the delay between retrying a failed message * @param currentAttempt How many attempts at handling the message have failed * @returns The number of milliseconds to delay retrying a failed message attempt */ calculateRetryDelay (currentAttempt: number): Milliseconds } ``` {% endcode %} An example of a retry strategy is as follows {% code title="default-retry-strategy.ts" %} ```typescript import { Milliseconds, RetryStrategy } from './retry-strategy' const MAX_DELAY_MS = 2.5 * 60 * 60 * 1000 // 2.5 hours const JITTER_PERCENT = 0.1 /** * A default message retry strategy that exponentially increases the delay between retries * from 5ms to 2.5 hrs for the first 10 attempts. Each retry delay includes a jitter of * up to 10% to avoid deadlock-related errors from continually blocking. */ export class DefaultRetryStrategy implements RetryStrategy { calculateRetryDelay (currentAttempt: number): Milliseconds { const numberOfFailures = currentAttempt + 1 const constantDelay: Milliseconds = Math.pow(5, numberOfFailures) const jitterAmount = Math.random() * JITTER_PERCENT * constantDelay const jitterDirection = Math.random() > 0.5 ? 1 : -1 const jitter = jitterAmount * jitterDirection const delay = Math.round(constantDelay + jitter) return Math.min(delay, MAX_DELAY_MS) } } ``` {% endcode %} # Dependency injection If your service uses a dependency injection provider and you want to use it to resolve class based handlers and workflows, you can provide an adapter to **@node-ts/bus-core** for it to use. This is provided as an adapter passed using **.withContainer()** to the configuration. This example uses [inversify](https://www.npmjs.com/package/inversify), but any IoC container that supports resolving instances via a class type will work. ```typescript import { Bus, ClassConstructor } from '@node-ts/bus-core' import { Container } from 'inversify' const start = async () => { const container = new Container() const bus = await Bus.configure() .withContainer({ get (type: ClassConstructor) { return container.get(type) } }) .initialize() } ``` # Long running processes ## Long running tasks Long running tasks are ones that cannot be completed within the normal message processing window. These are tasks like encoding a video or preparing a large file for download. Because they take a long time to complete, they should be run in a way that doesn't block the message handling process.‌ When a message is read from the underlying queue it must be processed and deleted within a configured timeout (usually a few minutes). If it is not deleted in this time, the underlying queue will assume that the consuming process has died and will flag the message as visible again and will be picked up by another handler that will do the same process. After a number of iterations like this, the message will be sent to the dead letter queue.‌ The other downside to running time-consuming processes in a handling window is that it blocks that worker from consuming other messages as it's waiting for the process to complete. Handlers should process messages as fast as possible.‌ ### Approaching long running tasks Let's say we have a command **EncodeVideo**. This might take up to an hour to complete so a handler can't wait for it to complete without the message being returned to the queue.‌ Instead a separate process should be started, and the events that follow reflect the asynchronous nature of the task.‌ ### A naive approach A simple way that is **not recommended** might be to background the task: ```typescript import { videoService } from 'services'​ export const encodeVideoHandler = (encodeVideo: EncodeVideo) => { setTimeout(async () => videoService.encode(encodeVideo), 0) } ``` This will background the task and the **EncodeVideo** command will be deleted, but it has a couple of flaws.‌ There is no retry mechanism at this point if the process fails. It will simply die without publishing any messages to indicate as such.‌ There's no effective load-balancing of tasks occurring. The same handler service may receive all of the **EncodeVideo** command and may attempt to have hundreds of these processes running in the background that eventually crash the instance.‌ Lastly if the service is restarted then the backgrounded tasks are killed and won't be retried.‌ ### A containerised approach If your app runs in kubernetes, docker swarm, ECS etc, then starting a pod/task per long running task can be very effective. This is outside the scope of what **@node-ts/bus** provides but can be implemented relatively simply with handlers.‌ When receiving an **EncodeVideo** command, use the handler to start the encoding process in a new pod/task and leave it up to the scheduler to place. This should also make it easier to scale out your app given the volume of these long running tasks being created.‌ #### Resiliency Just using handlers is a good start, but it won't provide the reliability needed if a task fails or gets terminated by the scheduler.‌ ​[Workflows](/guide/workflows) can be used to listen for [system messages](/guide/messages/system-messages) from the scheduler that indicate when a task as exited, and rerun the task if necessary.