Events
Connectum EventBus provides event-driven communication between microservices with proto-first routing, pluggable broker adapters, and a composable middleware pipeline.
This guide targets the 1.3.x release line. Features marked Since 1.3.0 require the corresponding package at 1.3.0 or later.
Architecture
Publishing skips the middleware chain: it only serializes the message and hands it to the adapter. Middleware wraps the consuming side only.
The EventBus sits between your service handlers and the message broker. It handles:
- Serialization -- automatically serializes/deserializes protobuf messages
- Routing -- maps proto service methods to topic subscriptions
- Middleware -- applies retry, DLQ, and custom middleware to every event
- Lifecycle -- manages adapter connect/disconnect with the server, graceful drain on shutdown
Core Concepts
Proto-First Routing
Event handlers are defined as proto services, mirroring ConnectRPC's ConnectRouter pattern. Each handler method receives a typed protobuf message and an EventContext:
// proto/orders/v1/events.proto
service OrderEventHandlers {
rpc OnOrderCreated(OrderCreated) returns (google.protobuf.Empty);
rpc OnOrderCancelled(OrderCancelled) returns (google.protobuf.Empty);
}import type { EventRoute } from '@connectum/events';
import { OrderEventHandlers } from '#gen/orders/v1/events_pb.ts';
const orderEvents: EventRoute = (events) => {
events.service(OrderEventHandlers, {
onOrderCreated: async (msg, ctx) => {
console.log(`Order ${msg.orderId} created`);
await ctx.ack();
},
onOrderCancelled: async (msg, ctx) => {
console.log(`Order ${msg.orderId} cancelled`);
await ctx.ack();
},
});
};Adapter Pattern
The EventAdapter interface abstracts away broker-specific details. Adapters handle connection management, message serialization at the wire level, and subscription lifecycle. Broker-specific configuration (credentials, tuning, stream names) is passed to the adapter constructor:
// NATS JetStream
import { NatsAdapter } from '@connectum/events-nats';
const adapter = NatsAdapter({ servers: 'nats://localhost:4222', stream: 'orders' });
// Kafka / Redpanda
import { KafkaAdapter } from '@connectum/events-kafka';
const adapter = KafkaAdapter({ brokers: ['localhost:9092'], clientId: 'my-service' });
// Redis Streams / Valkey
import { RedisAdapter } from '@connectum/events-redis';
const adapter = RedisAdapter({ url: 'redis://localhost:6379' });
// AMQP / RabbitMQ / LavinMQ
import { AmqpAdapter } from '@connectum/events-amqp';
const adapter = AmqpAdapter({ url: 'amqp://localhost:5672' });
// In-memory (testing)
import { MemoryAdapter } from '@connectum/events';
const adapter = MemoryAdapter();Middleware Pipeline
Middleware wraps event handlers in an onion model. Built-in middleware provides retry with configurable backoff and dead letter queue routing. See the middleware pipeline diagram for execution and error flow.
A delivery whose event type matches no route is acknowledged without running any handler or middleware, and it is not copied to the DLQ. See Unmatched events.
Each middleware receives the raw event, the event context, and a next() function to call the inner handler.
EventContext
Every event handler receives an EventContext with explicit acknowledgment control:
| Property | Description |
|---|---|
eventId | Unique event identifier |
eventType | Topic / event type name |
publishedAt | Publish timestamp |
attempt | Delivery attempt number (1-based) |
metadata | Event headers as ReadonlyMap<string, string> |
signal | AbortSignal -- aborted on server shutdown |
ack() | Acknowledge successful processing |
nack(requeue?) | Negative acknowledge -- request redelivery |
publish(schema, data, options?) | Publish on behalf of this delivery; still admitted while the bus stops, see Relay during shutdown |
Both ack() and nack() are idempotent -- calling either multiple times after the first call has no effect.
Shutdown Drain
A bus that was started with a shutdown signal (createServer({ eventBus }) passes the server's) closes its subscriptions as soon as that signal aborts, before stop() runs: a delivery that arrives afterwards is returned to the broker, not handled, and handlers already running are left alone (since 1.3.0, see the migration note).
eventBus.stop() closes every subscription, so no new events are delivered, and drains outstanding work at the same time. It disconnects the adapter once both have finished. Closing a subscription on the Redis, Kafka, and NATS adapters waits for the handler that is running, so the handler drain does not wait for the close: drainTimeout aborts a stuck handler, and that is what lets the close finish. Three budgets control the drain:
| Option | Default | Waits for | When the budget runs out |
|---|---|---|---|
drainTimeout | 30000 ms | Event handlers that are still running | Remaining handlers are aborted through their AbortSignal (0 aborts them immediately), and the bus then waits for them to return. A finite number from 0 to 2147483647; NaN, a negative number, Infinity, a larger number or a non-number makes createEventBus throw |
drainAbortGrace | None (unbounded) | Handlers that are still running after the abort | Since 1.3.0. The handlers still running are abandoned: stop() disconnects the adapter without waiting for the adapter to close the subscription of a handler that is still running, and the observer is told. Their later acknowledgement may fail; whether the message is delivered again depends on the broker and on whether the handler had already acknowledged it. 0 abandons at once after the abort. Must be finite |
drainPublishTimeout | Off | publish() calls that were already pending when stop() began | Nothing is aborted; the bus stops waiting and disconnects the adapter. The bus does not settle an unfinished publish() itself: the caller's promise settles later or stays pending, as the adapter decides |
Without drainAbortGrace, one handler that ignores its abort signal holds stop() forever. There is no default number because the time handlers need after an abort has not been measured; choose one from what your handlers do, and compare the total with your orchestrator's grace period: shutdown.timeout (transport phase) + drainTimeout + drainAbortGrace. The handler drain and the publish drain run at the same time, so the longer of them counts, not their sum.
observer receives what happened to work during the shutdown, and a dead-letter publish that failed at any time. It is called synchronously by the code that reports, and an error it throws (or a promise it rejects) is dropped, because reporting must never prevent the adapter from being disconnected or change what happens to the event.
| Event | When |
|---|---|
{ type: 'abandoned', handlers, publishes } | Once per stop(), only when something was left: handlers still running after drainAbortGrace, publishes still unsettled after their budget |
{ type: 'admitted', topic } | A ctx.publish made while the bus stops was admitted |
{ type: 'rejected', reason } | A ctx.publish made while the bus stops was refused: aborted (the drain was aborted) or not-in-flight (its handler had returned, or it belongs to an earlier start/stop cycle) |
{ type: 'dlq-republish', topic } | The built-in DLQ middleware republished a failed event while the bus stops |
{ type: 'dlq-failed', topic, eventType, eventId, error } | The built-in DLQ middleware could not publish a failed event to topic, at any time. The original is nacked without requeue; see DLQ middleware |
Without drainPublishTimeout, a pending publish() races the adapter disconnect, and a broker confirmation can fail because the connection closed first. Set the option to give those publishes time to settle:
const eventBus = createEventBus({
adapter,
routes: [orderEvents],
drainTimeout: 15_000,
drainPublishTimeout: 5_000,
});Both drains run at the same time, so the bus waits for the longer budget, not their sum. A handler that ignores its abort signal extends the handler drain until it returns. The publish drain is available since 1.3.0 and works with every adapter. undefined, 0, and negative values disable it. Tracking a publish for the drain adds no unhandled rejection of its own; each caller still receives the result of its publish() promise and must handle it.
The publish drain does not cover bus.publish() calls made after stop() begins. They reject immediately with an error that says the bus is stopping, including calls from handlers that are still draining. A handler that must hand its result on during shutdown publishes through ctx.publish instead; see Relay during shutdown. The dead letter republish of the DLQ middleware made while the bus stops is awaited before the adapter is disconnected, within drainAbortGrace when it is set.
Two stop() calls, also when one is made while start() is still running, share one shutdown and disconnect the adapter once.
createBroadcastSubscribers accepts the same drainTimeout, drainAbortGrace, drainPublishTimeout and observer options and passes them to every bus it creates.
When the bus is passed to createServer({ eventBus }), the server stops it in a shutdown hook. Hooks run after the server's shutdown.timeout phase, so drain time adds to total shutdown time. Keep the sum below your orchestrator's grace period; see Graceful shutdown. Exact option types are in EventBusOptions.
Adapter Comparison
Use the canonical Event Adapter selection matrix for broker trade-offs and direct links to exact adapter options. This overview owns the EventBus mental model; it does not duplicate broker configuration.
When to Use Events
| Pattern | Use Case | Transport |
|---|---|---|
| Request-response | Synchronous queries, CRUD operations | gRPC / ConnectRPC |
| Pub/sub events | Decoupled notifications, saga orchestration | EventBus |
| Streaming | Real-time data feeds, change data capture | gRPC server streaming |
Use EventBus when services need to react to events asynchronously without direct coupling. For synchronous communication, use Service Communication with gRPC clients.
Learn More
- Getting Started -- step-by-step setup tutorial
- Custom Topics -- proto options for topic naming
- Middleware -- retry, DLQ, custom middleware
- Adapters -- detailed adapter comparison and configuration
- @connectum/events -- Package Guide
- @connectum/events API -- Full API Reference