Send commands, publish events and run workflows over RabbitMQ or Amazon SQS. Failed messages are retried, nothing is lost, and your handlers stay plain functions.
npm i @node-ts/bus-core @node-ts/bus-messages
Messages are classes or plain definitions. Handlers get the message, its attributes and a context to send and publish through. Workflows react to messages over time and keep their state in a database.
import { Command, Event } from '@node-ts/bus-messages'
export class ReserveRoom extends Command {
static NAME = 'reservations/reserve-room'
$name = ReserveRoom.NAME
$version = 0
constructor(
readonly roomId: string,
readonly bookingId: string
) {
super()
}
}
export class RoomReserved extends Event {
static NAME = 'reservations/room-reserved'
$name = RoomReserved.NAME
$version = 0
constructor(
readonly roomId: string,
readonly bookingId: string
) {
super()
}
}import { handlerFor } from '@node-ts/bus-core'
import { ReserveRoom, RoomReserved } from '../messages'
import { reservationService } from '../services'
export const reserveRoomHandler = handlerFor(
ReserveRoom,
async (command, _attributes, ctx) => {
await reservationService.reserveRoom(command.roomId, command.bookingId)
// Published once the handler resolves, and dropped if it throws
await ctx.publish(new RoomReserved(command.roomId, command.bookingId))
}
)import { Bus } from '@node-ts/bus-core'
import { reserveRoomHandler } from './handlers/reserve-room-handler'
import { messageTypes } from './message-types.generated'
import { ReserveRoom } from './messages'
const bus = Bus.configure()
.withMessageTypes(messageTypes)
.withHandler(reserveRoomHandler)
.build()
await bus.initialize()
// Start the bus to begin handling messages
await bus.start()
await bus.send(
new ReserveRoom(
'63a65cf0-d239-4b83-96da-f33f013db23a',
'12b85a56-e929-47a8-9ac3-e87739d5d215'
)
)import { defineWorkflow } from '@node-ts/bus-core'
import {
EmailReceipt,
ItemPurchased,
ItemShipped,
ReceiptEmailed,
ShipItem
} from '../messages'
import { FulfilmentWorkflowState } from './fulfilment-workflow-state'
export const fulfilmentWorkflow = defineWorkflow(FulfilmentWorkflowState)
// A purchase starts a new workflow, which ships the item
.startedBy(ItemPurchased, async ({ itemId, customerId }, _state, ctx) => {
await ctx.send(new ShipItem(itemId, customerId))
return { itemId, customerId, status: 'shipping-item' as const }
})
// ItemShipped is published by the ShipItem handler, so it carries this
// workflow's id and is routed back to this instance
.when(ItemShipped, async (event, { itemId, customerId }, ctx) => {
await ctx.send(new EmailReceipt(itemId, customerId))
return { status: 'emailing-receipt' as const, shippedAt: event.shippedAt }
})
.when(ReceiptEmailed, (_event, _state, ctx) =>
ctx.complete({ status: 'complete' })
)Write messages and handlers once, and pick a transport for the queues and a persistence for workflow state. The in-memory defaults need nothing to run, for development and tests.
Message-based systems are known for reliability, resilience and throughput. They scale out easily, survive outages without losing data, and recover once the outage is over. @node-ts/bus sets up the transport, routes messages to handlers, propagates attributes and retries failures, so more of your code is about your application.