Migrating from v2

Overview

The legacy service-connect v2.x package and the new @serviceconnect/* v3 packages share the wire format but otherwise look very different. v2 was a single CommonJS package with a class-based API and AMQP-only transport. v3 is an ESM monorepo with a typed bus, pluggable transport, and first-class patterns for everything v2 only had on the TODO list.

The good news is that the on-the-wire envelope is unchanged. v2 publishes JSON payloads with MessageType, RequestMessageId, ResponseMessageId, SourceAddress, and the rest of the PascalCase header set; v3 reads and writes the same headers. A v3 consumer can take over a v2 producer’s traffic without any broker reconfiguration, and v2 consumers can keep reading messages a v3 publisher emits. This lets you migrate services one at a time rather than all at once.

The rest of this page walks every public concept in v2 and shows the v3 equivalent. Each section has a v2 snippet on the left and the v3 replacement on the right; where v3 introduces a new concept, the prose calls out what changed and why.

1. Lifecycle

v2 was class-based. You constructed a Bus with the full configuration in one object, then called init() to connect, and close() to shut down.

// v2
import { Bus } from 'service-connect';

const bus = new Bus({
  amqpSettings: {
    queue: { name: 'orders' },
    host: 'amqp://localhost',
  },
});

await bus.init();
// ... use bus ...
await bus.close();

v3 swaps the class for a factory. The bus is constructed by createBus, the transport is constructed by createRabbitMQTransport (or rabbitMQWithRegistry if you want polymorphism), and the bus is AsyncDisposable so await using cleans it up automatically.

// v3
import { createBus } from '@serviceconnect/core';
import { createRabbitMQTransport } from '@serviceconnect/rabbitmq';

await using bus = createBus({
  queue: { name: 'orders' },
  transport: createRabbitMQTransport({ url: 'amqp://localhost' }),
});

await bus.start();
// ... use bus ...
// `await using` calls bus.stop() automatically.

If you do not want await using, call await bus.stop() explicitly. stop() drains in-flight work before closing the transport. (It accepts an optional AbortSignal parameter, but that parameter is currently ignored and has no effect on the drain or teardown.)

2. Registration

v2 had a single registration method, addHandler. Handlers received four positional arguments: the message, the headers, the type string, and a reply callback.

// v2
await bus.addHandler('OrderPlaced', async (message, headers, type, replyCallback) => {
  console.log('placed', message.orderId);
  await replyCallback('OrderConfirmed', { orderId: message.orderId });
});

v3 splits registration into two steps. registerMessage<T>(typeName) declares the type to the registry (this is what lets v3 do typed polymorphism). handle<T>(typeName, fn) attaches a handler. Both calls return the bus, so they chain. The handler receives the typed payload and a ConsumeContext — the headers, type, and reply primitive all live on the context.

// v3
interface OrderPlaced {
  correlationId: string;
  orderId: string;
}

interface OrderConfirmed {
  correlationId: string;
  orderId: string;
}

bus
  .registerMessage<OrderPlaced>('OrderPlaced')
  .registerMessage<OrderConfirmed>('OrderConfirmed')
  .handle<OrderPlaced>('OrderPlaced', async (msg, ctx) => {
    console.log('placed', msg.orderId);
    await ctx.reply<OrderConfirmed>('OrderConfirmed', {
      correlationId: msg.correlationId,
      orderId: msg.orderId,
    });
  });

Differences worth calling out:

The handler signature shrinks from four positional arguments to (message, context). Everything else moves onto context: context.headers, context.messageType, context.headers.sourceAddress, context.reply(...), context.signal. (The source address lives on context.headers.sourceAddress; there is no top-level context.sourceAddress.)
Every v3 message must extend Message, which means it must carry a correlationId: string. v2 did not require this — adding it is the main schema change you will see at every call site.
v3 is strict about registration: bus.handle('X', ...) throws at startup if 'X' was not declared with registerMessage. v2 would happily accept any string.

3. Send and publish

v2 took the endpoint as the first positional argument; v3 moves it into the options object. v2 also accepted an array for fan-out send; v3 splits that into a separate method, sendToMany, to keep the single-endpoint signature obvious.

// v2 — send
await bus.send('orders', 'OrderPlaced', { orderId: 'o-1' });
await bus.send(['orders', 'fulfilment'], 'OrderPlaced', { orderId: 'o-1' });

// v2 — publish
await bus.publish('OrderPlaced', { orderId: 'o-1' });

// v3 — send
await bus.send<OrderPlaced>('OrderPlaced', payload, { endpoint: 'orders' });
await bus.sendToMany<OrderPlaced>('OrderPlaced', payload, ['orders', 'fulfilment']);

// v3 — publish
await bus.publish<OrderPlaced>('OrderPlaced', payload);

Both methods take an optional headers field on the options object for custom AMQP headers. The bus owns the standard envelope headers (RequestMessageId, SourceAddress, etc.) and you should not set those manually.

4. Request / reply

v2’s request/reply was callback-driven and could fire the callback once per reply for scatter-gather. The caller passed a callback as the fourth argument; v2 wired up the response correlation and invoked the callback for each matching reply.

// v2 — single-endpoint request
await bus.sendRequest(
  'pricing',
  'GetQuote',
  { sku: 'abc' },
  async (reply, headers, type) => {
    console.log('quote', reply.amount);
  },
);

// v2 — broadcast request, expect 3 replies, 5-second timeout
await bus.publishRequest(
  'GetQuote',
  { sku: 'abc' },
  async (reply) => {
    console.log('quote', reply.amount);
  },
  3,
  5000,
);

v3 returns promises. sendRequest resolves to the first reply; sendRequestMulti resolves to an array. Cancellation is via AbortSignal. A RequestTimeoutError is thrown if timeoutMs elapses before the expected reply count is reached.

// v3 — single-endpoint request
const quote = await bus.sendRequest<GetQuote, Quote>(
  'GetQuote',
  { correlationId: 'c-1', sku: 'abc' },
  { endpoint: 'pricing', timeoutMs: 5000 },
);
console.log('quote', quote.amount);

// v3 — gather N replies from competing consumers on one endpoint
const quotes = await bus.sendRequestMulti<GetQuote, Quote>(
  'GetQuote',
  { correlationId: 'c-1', sku: 'abc' },
  { endpoint: 'pricing', expectedReplyCount: 3, timeoutMs: 5000 },
);
console.log(quotes.length, 'quotes received');

For the broadcast (publish-then-collect-replies) case, v3 exposes bus.publishRequest for symmetry with v2’s broadcast request — it takes a per-reply callback and an optional expectedReplyCount. Use it when responders live on different queues (true scatter-gather across heterogeneous services); use sendRequestMulti when responders are competing consumers on the same queue. The promise-returning sendRequest / sendRequestMulti compose with await, Promise.all, and AbortController in ways the callback form cannot.

5. Transport

v2 hard-coded RabbitMQ. The broker URL, queue name, SSL, prefetch, retry, error queue, audit queue — everything lived under amqpSettings on the bus config. Switching transports would have required forking the library.

// v2
const bus = new Bus({
  amqpSettings: {
    queue: { name: 'orders' },
    host: 'amqp://user:pass@broker:5672/vhost',
    ssl: { enabled: true },
    prefetch: 50,
    maxRetries: 5,
    errorQueue: 'orders.errors',
    auditEnabled: true,
    auditQueue: 'orders.audit',
  },
});

v3 splits the bus and the transport. The transport is a separate package (@serviceconnect/rabbitmq) that implements ITransportProducer + ITransportConsumer from @serviceconnect/core. The bus does not know or care which transport you pass it, which is what makes the package set extensible.

// v3
import { createBus } from '@serviceconnect/core';
import { createRabbitMQTransport } from '@serviceconnect/rabbitmq';

const transport = createRabbitMQTransport({
  url: 'amqps://user:pass@broker:5671/vhost',
  consumer: {
    prefetch: 50,
    maxRetries: 5,
    errorQueue: 'orders.errors',
    auditEnabled: true,
    auditQueue: 'orders.audit',
  },
});

const bus = createBus({ queue: { name: 'orders' }, transport });

If you need polymorphic dispatch — a publish of a derived type also reaching base-type subscribers — use rabbitMQWithRegistry({ url }, registry) instead. It wires the registry’s parent map into the transport so the producer publishes each derived message to its own exchange and every declared parent’s exchange.

6. Patterns built in v3

v2 supported point-to-point, pub/sub, request/reply, scatter-gather, retries, auditing, error handling, SSL, and priority queues. The README also listed a “TODO” set of patterns that were never built. v3 ships all of them.

Process Manager (saga) — v2 TODO. v3 has ProcessHandler<TData, TMessage> and a fluent DSL: bus.registerProcessData<TData>('DataType').registerProcess('order', { store, timeoutStore }).startsWith<M>('OrderPlaced', handler).handles<M>('PaymentReceived', handler). See Process Manager.
Aggregator — v2 TODO. v3 has the Aggregator<T> abstract class with batchSize/timeout/execute(batch), registered via bus.registerAggregator(type, instance, { store }). See Aggregator.
Routing Slip — v2 TODO. v3 has bus.route(typeName, msg, [...destinations]) and a RoutingSlip header that each handler forwards along the slip until it is empty. See Routing Slip.
Streaming — v2 TODO. v3 has bus.openStream(endpoint, typeName) returning a StreamSender<T>; consumers register with bus.handleStream(typeName, async (stream) => { for await (const chunk of stream) { ... } }). See Streaming.
Polymorphic dispatch — v2 TODO. v3 supports declared parents via registry.register<T>(name, { parents: ['Parent'] }). A handler for the parent type runs for every derived child. See Polymorphic Messages.
Filters as first-class registration — v2 had a filters.before/filters.after/filters.outgoing array on the config. v3 has bus.use(stage, asFilter(fn), asMiddleware(fn)) across four stages: 'outgoing', 'beforeConsuming', 'afterConsuming', and 'onConsumedSuccessfully'. See Filters.
Content-based routing — not in v2. v3 supports it directly as a bus.handle pattern that calls bus.send based on payload contents. See Content-Based Routing.
Scatter-gather — v2 had it via publishRequest. v3 has both publishRequest and the more ergonomic sendRequestMulti. See Scatter-Gather.

7. Module system

v2 was CommonJS. The package set "main": "index.js" and shipped a .d.ts next to it; consumers used const { Bus } = require('service-connect') (or import { Bus } from 'service-connect' if their bundler did interop).

// v2
const { Bus } = require('service-connect');
// or, with esModuleInterop:
import { Bus } from 'service-connect';

v3 is ESM-only. Every @serviceconnect/* package sets "type": "module" and exports .js files with "exports" maps targeting Node’s ES Module loader. CommonJS consumers cannot require() v3; they must switch to import, or use dynamic import() from inside a CJS module.

// v3
import { createBus } from '@serviceconnect/core';
import { createRabbitMQTransport } from '@serviceconnect/rabbitmq';

In TypeScript you must also bump tsconfig.json to use NodeNext module resolution, which is the only resolver that understands the "exports" map:

{
  "compilerOptions": {
    "module": "nodenext",
    "moduleResolution": "nodenext",
    "target": "es2022"
  }
}

If your application is currently CommonJS, the smallest change is to set "type": "module" on your own package.json and convert your entry points. The TypeScript compiler will guide you through the rest (rename .ts to .mts if you have a mix, add .js to relative import specifiers).

8. Engine support

v2 required Node 18+. v3 requires Node 22 LTS. The headline reasons:

Top-level await — every v3 example uses it. It works in Node 18, but the v3 packages rely on it inside their own ESM entry points.
Native AsyncDisposable — await using bus = createBus(...) is a Node 22 language feature (TypeScript 5.2+). v3 implements [Symbol.asyncDispose]() on the bus so await using calls bus.stop() automatically when the scope exits.
Web Streams APIs — the streaming pattern uses ReadableStream/WritableStream from the global scope, which is most stable on Node 22.
Native test runner improvements — not a hard dependency, but the test suite uses Node 22 features. If you depend on the v3 packages in a long-lived service, staying on Node 22 LTS keeps you on a supported runtime through 2027.

9. Configuration mapping

The table maps every v2 default in lib/settings.js to its v3 equivalent. Anything missing on either side is called out.

v2 key	v3 equivalent
`amqpSettings.queue.name`	`BusOptions.queue.name`
`amqpSettings.queue.durable`	Fixed at `true`; not configurable. `queueArguments` only feeds the AMQP `arguments` (x-args) map and cannot change the queue’s `durable` declaration flag.
`amqpSettings.queue.exclusive`	Not exposed; v3 queues are non-exclusive by design.
`amqpSettings.queue.autoDelete`	Not exposed; v3 queues are not auto-delete. `queueArguments` only feeds the AMQP `arguments` map and cannot set the `autoDelete` declaration flag.
`amqpSettings.queue.noAck`	Not exposed; v3 always uses explicit ack.
`amqpSettings.queue.maxPriority`	Set `'x-max-priority'` in `transport.consumer.queueArguments`.
`amqpSettings.host`	`transport: createRabbitMQTransport({ url: 'amqp://...' })`.
`amqpSettings.ssl.enabled`	Set `tls: true` on `RabbitMQTransportOptions` (equivalent to using an `amqps://` URL).
`amqpSettings.ssl.key` / `cert` / `ca` / `pfx`	Pass them via `RabbitMQTransportOptions.tls` as a `node:tls` `ConnectionOptions` object (`{ ca, cert, key, pfx }`); the transport forwards it to `rabbitmq-client`. `tls` accepts `boolean
`amqpSettings.retryDelay`	`transport.consumer.retryDelay` (default `3000`).
`amqpSettings.maxRetries`	`transport.consumer.maxRetries` (default `3`).
`amqpSettings.errorQueue`	`transport.consumer.errorQueue` (default `'errors'`; set `null` to disable).
`amqpSettings.auditQueue`	`transport.consumer.auditQueue` (default `'audit'`).
`amqpSettings.auditEnabled`	`transport.consumer.auditEnabled` (default `false`).
`amqpSettings.prefetch`	`transport.consumer.prefetch` (default `100`).
`filters.outgoing`	`bus.use('outgoing', asFilter(fn))`.
`filters.before`	`bus.use('beforeConsuming', asFilter(fn))`.
`filters.after`	`bus.use('afterConsuming', asFilter(fn))`.
`handlers` (registered up-front in config)	`bus.handle<T>(typeName, fn)` per type. Polymorphic parents declared via the registry.
`client` (custom transport class)	Pass any `{ producer, consumer }` pair conforming to `ITransportProducer` + `ITransportConsumer`.
`logger`	`BusOptions.logger`. The v3 `Logger` has six methods — `trace`, `debug`, `info`, `warn`, `error`, `fatal` — each `(msg: string, meta?: object)`; only `child` is optional.

v3-only configuration without a v2 equivalent:

BusOptions.serializer — pluggable serializer (JSON is the default).
BusOptions.defaultRequestTimeout — currently inert/reserved: it is accepted but never read. sendRequest/sendRequestMulti require an explicit positive per-call timeoutMs (they throw ArgumentOutOfRangeError otherwise) and do not fall back to this option.
BusOptions.timeoutPollIntervalMs — request-reply timeout poll cadence.
BusOptions.aggregatorFlushIntervalMs — aggregator flush timer cadence.
BusOptions.consumeWrapper — wrap every handler invocation (e.g. for OpenTelemetry).
transport.connectionName — surfaces in the RabbitMQ management UI.
transport.producer.publishConfirmTimeoutMs / maxAttempts / maxMessageSize.

10. Filters and middleware

v2 ran two filter arrays on every inbound message (filters.before, then handlers, then filters.after) and one on every outbound message (filters.outgoing). Filters returned false to short-circuit.

// v2
const bus = new Bus({
  // ...
  filters: {
    outgoing: [async (message, headers, type, bus) => {
      headers['X-Tenant'] = 'acme';
    }],
    before: [async (message, headers, type, bus) => {
      if (!message.tenant) return false; // drop messages without a tenant
    }],
    after: [async (message, headers, type, bus) => {
      console.log('processed', type);
    }],
  },
});

v3 has four stages — 'outgoing', 'beforeConsuming', 'afterConsuming', and 'onConsumedSuccessfully' — and splits the pipeline into two kinds of unit:

Filters decide whether processing continues. They return FilterAction.Continue or FilterAction.Stop. Stop short-circuits the stage. Use asFilter(fn) to register one — fn must return a FilterAction (return FilterAction.Continue explicitly on the non-Stop path). A filter receives the raw Envelope ({ headers, body }), not a decoded message.
Middleware wraps the rest of the pipeline. It takes a next() callback and can run code before and after. Use asMiddleware(fn) to register one.

// v3
import { asFilter, asMiddleware, FilterAction } from '@serviceconnect/core';

bus
  .use(
    'outgoing',
    asMiddleware(async (context, next) => {
      context.envelope.headers['X-Tenant'] = 'acme';
      await next();
    }),
  )
  .use(
    'beforeConsuming',
    asFilter(async (envelope) => {
      // Filters run before deserialization, so they see only the raw
      // headers/body — there is no decoded message here. Inspect a header.
      if (!envelope.headers['X-Tenant']) return FilterAction.Stop;
      return FilterAction.Continue;
    }),
  )
  .use(
    'onConsumedSuccessfully',
    asMiddleware(async (context, next) => {
      await next();
      console.log('processed', context.envelope.headers.messageType);
    }),
  );

The 'onConsumedSuccessfully' stage is new in v3; it runs after a handler resolves successfully and is the natural home for ack-side bookkeeping (acknowledging an idempotency token, emitting a “consumed” trace event, etc.).

11. Wildcard handlers

v2 supported bus.addHandler('*', cb) as a catch-all. The wildcard handler ran for every type, alongside (not instead of) any type-specific handler.

// v2
await bus.addHandler('*', async (message, headers, type) => {
  console.log('saw', type);
});

v3 has no literal wildcard. The same effect comes from polymorphic dispatch — declare a base type, mark every concrete type as descending from it via parents, and handle the base type. The dispatch system runs the base handler for every descendant.

// v3
interface BaseMessage extends Message {
  tenant: string;
}

interface OrderPlaced extends BaseMessage {
  orderId: string;
}

interface PaymentReceived extends BaseMessage {
  paymentId: string;
}

const registry = createMessageTypeRegistry();
registry.register<BaseMessage>('BaseMessage');
registry.register<OrderPlaced>('OrderPlaced', { parents: ['BaseMessage'] });
registry.register<PaymentReceived>('PaymentReceived', { parents: ['BaseMessage'] });

bus
  .registerMessage<BaseMessage>('BaseMessage')
  .registerMessage<OrderPlaced>('OrderPlaced')
  .registerMessage<PaymentReceived>('PaymentReceived')
  .handle<BaseMessage>('BaseMessage', async (msg, ctx) => {
    console.log('saw', ctx.messageType, msg.tenant);
  });

The base handler now runs for every OrderPlaced and PaymentReceived. The trade-off is that types and parents are declared up-front instead of inferred from a literal '*' — but you get typed access to whatever fields the base type promises (here, tenant).

12. Reply primitive

v2’s reply primitive was the fourth positional argument to a handler. It took the reply’s type string and payload; the bus wired up ResponseMessageId from the incoming RequestMessageId and sent to the inbound SourceAddress.

// v2
await bus.addHandler('GetQuote', async (message, headers, type, replyCallback) => {
  await replyCallback('Quote', { amount: 42, sku: message.sku });
});

v3 moves the reply primitive onto ConsumeContext as ctx.reply<TReply>(typeName, payload, options?). It is typed and async. The bus correlates and routes the reply the same way v2 did; the difference is in how it surfaces.

// v3
bus.handle<GetQuote>('GetQuote', async (msg, ctx) => {
  await ctx.reply<Quote>('Quote', {
    correlationId: msg.correlationId,
    amount: 42,
    sku: msg.sku,
  });
});

One behavioural change: v3 throws if you call ctx.reply from a handler whose incoming envelope has no sourceAddress header. v2 silently delivered the reply to the wrong place (or nowhere) in that case. The new behaviour fails fast — replies belong on point-to-point flows, not on broadcasts you observed by accident.