Request/Reply

Overview

Request/Reply turns the bus into a typed RPC channel: the caller sends a request, awaits a single reply, and gets the response back as a strongly-typed value. Under the hood it is still asynchronous messaging — there is no open socket, no blocked thread — but the API surface (await bus.sendRequest(...)) reads like a normal function call.

When to use

Reach for Request/Reply when the caller genuinely needs an answer before it can continue: pricing lookups, validation checks, on-demand projections of remote state. The fact that the response arrives over the bus rather than HTTP is invisible to the caller, but it buys you transport-level retries, broker-side durability for the request, and uniform observability with the rest of your messages.

It is also useful as a migration ramp when you are pulling a synchronous HTTP API onto an async backbone. Call sites stay shaped the same; only the transport changes.

When not to use

Long-running operations are another bad fit. If the responder takes minutes, the request becomes a saga — model it as a conversation of events rather than blocking on a single reply.

Example

interface PriceQuoteRequest extends Message {
  sku: string;
  qty: number;
}
interface PriceQuoteReply extends Message {
  unitPrice: number;
}

const reply = await bus.sendRequest<PriceQuoteRequest, PriceQuoteReply>(
  'PriceQuoteRequest',
  { correlationId: 'c-1', sku: 'SKU-1', qty: 3 },
  { endpoint: 'pricing', timeoutMs: 5_000 },
);

console.log('unit price', reply.unitPrice);

The responder side is an ordinary handler that calls ctx.reply(...) with the typed payload. The bus on each side handles correlation, the reply queue, and the deserialisation — your handler never sees the plumbing.

Behind the scenes

Each request gets a fresh requestMessageId generated by the request-reply manager (via newMessageId()). That id is written into the outbound envelope headers; the responder copies it back onto every reply it emits via ctx.reply, where it travels as the responseMessageId header. It is what makes simultaneous in-flight requests safe — the bus uses it to route the right reply back to the right awaiter. (correlationId is your own application-level field and plays no part in this routing.)

The requester owns a private reply queue, declared once at bus start. The reply-queue name is included in the outbound envelope so the responder knows where to address its reply. Replies arriving on that queue are demultiplexed by their responseMessageId header — matched back to the pending request whose requestMessageId it carries — and used to resolve the pending promise.

Timeouts raise RequestTimeoutError after timeoutMs has elapsed. sendRequest requires a positive timeoutMs — there is no implicit default, and an absent or non-positive value throws ArgumentOutOfRangeError before the request is sent. At that point the pending promise is rejected and the correlation entry is dropped; a late-arriving reply with that id is discarded silently.

Cancellation is cooperative via AbortSignal. Pass { signal } in the options and any abort — whether the signal is already aborted before the request is published or fires while the request is in flight — rejects the promise with AbortError. (RequestSendCancelledError is a separate error, raised only when the underlying send fails before reaching the broker — a transport failure, not a cancellation.) The framework does not attempt to cancel the responder — it cannot. Cancellation only releases the caller.