Appearance

GuidesDocs 1.3.x

Getting Started with Events ​

This guide targets the documented 1.3.x release line. The drainPublishTimeout option in the example requires @connectum/events 1.3.0 or later; it is unavailable from the published 1.2.x package until 1.3.0 is released.

This guide walks you through setting up event-driven communication between Connectum microservices using the EventBus. You finish with one published event handled successfully; broker selection and middleware tuning are optional next steps.

Prerequisites ​

  • A working Connectum project with @connectum/core
  • Proto tooling configured (buf for protobuf generation)
  • A message broker (or use MemoryAdapter for local development)

The snippets use TypeScript source directly, with import_extension=.ts in buf.gen.yaml and a #gen/* import alias for generated files, as in the Quickstart. If your project compiles before execution, follow its configured import extension instead.

Step 1: Install Packages ​

Install @connectum/events plus the adapter for your broker:

BrokerAdapter package
NATS JetStream@connectum/events-nats
Kafka / Redpanda@connectum/events-kafka
Redis Streams / Valkey@connectum/events-redis
AMQP / RabbitMQ@connectum/events-amqp
Memory (testing)built in -- no extra package

The commands below install the NATS adapter; substitute the package from the table for another broker, or drop it entirely to start with the in-memory adapter.

bash
npm install @connectum/events @connectum/events-nats
bash
pnpm add @connectum/events @connectum/events-nats
bash
bun add @connectum/events @connectum/events-nats

Step 2: Define Proto Event Handlers ​

Create a proto service that defines your event handlers. Each RPC method represents a handler for one event type. The input message is the event payload; the return type should be google.protobuf.Empty.

protobuf
// proto/notifications/v1/events.proto
syntax = "proto3";

package notifications.v1;

import "google/protobuf/empty.proto";

// Events published by the User service
message UserCreated {
  string id = 1;
  string email = 2;
  string name = 3;
}

message UserDeleted {
  string id = 1;
}

// Event handler service for the Notification service
service NotificationEventHandlers {
  rpc OnUserCreated(UserCreated) returns (google.protobuf.Empty);
  rpc OnUserDeleted(UserDeleted) returns (google.protobuf.Empty);
}

Generate the TypeScript code using your project's protobuf generation script:

bash
npm run build:proto
bash
pnpm run build:proto
bash
bun run build:proto

Step 3: Create an Adapter ​

Create the adapter instance with broker-specific configuration:

typescript
import { NatsAdapter } from '@connectum/events-nats';

const adapter = NatsAdapter({
  servers: process.env.NATS_URL ?? 'nats://localhost:4222',
  stream: 'notifications',
});
typescript
import { KafkaAdapter } from '@connectum/events-kafka';

const brokers = (process.env.KAFKA_BROKERS ?? 'localhost:9092').split(',');
const adapter = KafkaAdapter({
  brokers,
  clientId: 'notification-service',
});
typescript
import { RedisAdapter } from '@connectum/events-redis';

const adapter = RedisAdapter({
  url: process.env.REDIS_URL ?? 'redis://localhost:6379',
});
typescript
import { AmqpAdapter } from '@connectum/events-amqp';

const adapter = AmqpAdapter({
  url: process.env.AMQP_URL ?? 'amqp://localhost:5672',
});
typescript
import { MemoryAdapter } from '@connectum/events';

const adapter = MemoryAdapter();

Step 4: Implement Event Handlers ​

Define typed event handlers using the EventRoute pattern. This mirrors ConnectRPC's router registration:

typescript
// src/events/notificationEvents.ts
import type { EventRoute } from '@connectum/events';
import { NotificationEventHandlers } from '#gen/notifications/v1/events_pb.ts';

export const notificationEvents: EventRoute = (events) => {
  events.service(NotificationEventHandlers, {
    onUserCreated: async (msg, ctx) => {
      console.log(`Sending welcome email to ${msg.email}`);
      // ... send email logic
      await ctx.ack();
    },

    onUserDeleted: async (msg, ctx) => {
      console.log(`Cleaning up notifications for user ${msg.id}`);
      // ... cleanup logic
      await ctx.ack();
    },
  });
};

Acknowledgment

Successful handler completion auto-acks the event. The Events concept guide owns acknowledgment and metadata semantics.

Step 5: Create the EventBus ​

Wire everything together with createEventBus():

typescript
// src/eventBus.ts
import { createEventBus } from '@connectum/events';
import { NatsAdapter } from '@connectum/events-nats';
import { notificationEvents } from './events/notificationEvents.ts';

const adapter = NatsAdapter({
  servers: process.env.NATS_URL ?? 'nats://localhost:4222',
  stream: 'notifications',
});

export const eventBus = createEventBus({
  adapter,
  routes: [notificationEvents],
  group: 'notification-service',
  handlerTimeout: 30_000,  // Per-event handler timeout (default: 30s)
  drainTimeout: 15_000,    // Wait up to 15s for in-flight handlers on shutdown
  drainPublishTimeout: 5_000, // Opt-in (default: off): wait up to 5s for in-flight publishes on shutdown
  middleware: {
    retry: { maxRetries: 3, backoff: 'exponential' },
    dlq: { topic: 'notification-service.dlq' },
  },
});

drainTimeout and drainPublishTimeout both apply when the bus stops. See Shutdown drain for what each budget waits for and how they combine.

Step 6: Integrate with Connectum Server ​

Pass the EventBus to createServer() for automatic lifecycle management:

typescript
// src/index.ts
import { createServer } from '@connectum/core';
import { Healthcheck, healthcheckManager, ServingStatus } from '@connectum/healthcheck';
import { eventBus } from './eventBus.ts';
import { routes } from './services/routes.ts';

const server = createServer({
  services: [routes],
  eventBus,
  protocols: [Healthcheck({ httpEnabled: true })],
  shutdown: { autoShutdown: true },
});

server.on('ready', () => {
  healthcheckManager.update(ServingStatus.SERVING);
  console.log(`Notification service ready on port ${server.address?.port}`);
});

await server.start();

The server now owns the EventBus lifecycle. See the Events concept guide for that lifecycle boundary.

Step 7: Publish Events ​

From another service (e.g., User Service), publish typed events:

typescript
import { createEventBus } from '@connectum/events';
import { NatsAdapter } from '@connectum/events-nats';
import { UserCreatedSchema } from '#gen/notifications/v1/events_pb.ts';

const eventBus = createEventBus({
  adapter: NatsAdapter({
    servers: process.env.NATS_URL ?? 'nats://localhost:4222',
    stream: 'notifications',
  }),
});

await eventBus.start();

// Publish a typed event -- serialized as protobuf, routed by schema.typeName
await eventBus.publish(UserCreatedSchema, {
  id: '123',
  email: '[email protected]',
  name: 'Alice',
});

await eventBus.stop();

The publisher and subscriber must use the same broker and stream. This publisher matches the NATS bus in Step 5; if you chose another adapter, configure both sides for that broker. The topic defaults to the message's typeName (e.g., notifications.v1.UserCreated). See Custom Topics to override this.

Verify the result ​

Start the notification service before running the publisher. Its handler should print Sending welcome email to [email protected]. A resolved publish() alone confirms the adapter's publish operation; verify the consumer output as well.

Troubleshooting ​

  • No consumer output: check that both processes use the same broker URL, adapter settings, and topic. With the NATS default deliverPolicy: 'new', start the subscriber before publishing to a new consumer.
  • The broker is unavailable: eventBus.start() fails or waits according to the selected adapter's connection policy; see Adapter selection.
  • Protobuf generation or imports fail: use the generation command and import extension configured by your project's buf.gen.yaml.

Full Working Example ​

Here is a minimal two-service setup using the MemoryAdapter for local testing:

typescript
import { createEventBus, MemoryAdapter } from '@connectum/events';
import type { EventRoute } from '@connectum/events';
import {
  NotificationEventHandlers,
  UserCreatedSchema,
} from '#gen/notifications/v1/events_pb.ts';

// Shared in-memory adapter (for testing only)
const adapter = MemoryAdapter();

// Consumer: Notification Service
const notificationEvents: EventRoute = (events) => {
  events.service(NotificationEventHandlers, {
    onUserCreated: async (msg, ctx) => {
      console.log(`Welcome email sent to ${msg.email}`);
      await ctx.ack();
    },
    onUserDeleted: async (msg, ctx) => {
      console.log(`Notifications cleaned for user ${msg.id}`);
      await ctx.ack();
    },
  });
};

const consumerBus = createEventBus({
  adapter,
  routes: [notificationEvents],
  group: 'notification-service',
});

// Producer: User Service
const producerBus = createEventBus({ adapter });

await consumerBus.start();
await producerBus.start();

// Publish — the consumer handler fires synchronously with MemoryAdapter
await producerBus.publish(UserCreatedSchema, {
  id: '1',
  email: '[email protected]',
  name: 'Alice',
});
// Output: "Welcome email sent to [email protected]"

await consumerBus.stop();
await producerBus.stop();

Next Steps ​