Broker TransportsSupported

Redis Pub/Sub, core NATS and RabbitMQ implement one public strategy contract, but they do not pretend to offer equivalent durability. Choose from the settlement matrix below before choosing from familiarity.

The adapters and peers are opt-in; yarn add @zmdb/core@1.0.0-beta.2 installs none of them. @zmdb/app/messaging owns the broker-neutral contract and application lifecycle, while each adapter owns only its protocol client, framing, subscriptions, replies, and settlement.

The strategy boundary#

A strategy owns broker framing, subscriptions, replies and settlement. The application owns payload validation, handler invocation and retry policy:

import type { TraceCarrier } from '@zmdb/app/observability';
import type { DispatchOutcome, MessageReply, RawMessage, TransportCapabilities, TransportRequest } from '@zmdb/app/messaging';

export interface TransportStrategy {
  readonly name: string;
  readonly capabilities: TransportCapabilities;
  listen(dispatch: (message: RawMessage) => Promise<DispatchOutcome>): Promise<void>;
  send(request: TransportRequest): Promise<MessageReply>;
  emit(pattern: string, payload: unknown, carrier?: TraceCarrier): Promise<void>;
  close(graceMs: number): Promise<void>;
}

The packaged strategies use a versioned JSON envelope and reject an undefined payload. Malformed JSON reaches the dispatcher as RawMessage.parseError, with the original input retained for onInvalidPayload. traceparent / tracestate travel in the envelope; correlation and reply destinations use either envelope fields or native broker metadata.

Settlement matrix#

Capability order is redelivery / deadLetter / requestResponse. Redis and core NATS therefore require dispatcher.onUndeliverable; RabbitMQ owns the retry and dead-letter topology described below.

Install only the selected client#

The neutral @zmdb/app/messaging entry imports no broker client. Install the optional peer alongside the adapter you use:

yarn add @zmdb/transport@1.0.0-beta.2 redis@^6.2.1
yarn add @zmdb/transport@1.0.0-beta.2 @nats-io/transport-node@^3.4.0
yarn add @zmdb/transport@1.0.0-beta.2 amqplib@^2.0.1

import { createNatsStrategy } from '@zmdb/transport/nats';
import { createRabbitMqStrategy } from '@zmdb/transport/rabbitmq';
import { createRedisStrategy } from '@zmdb/transport/redis';

Kafka events#

Install @zmdb/transport/kafka with kafkajs@>=2.2.4 <3.0.0 and attach createKafkaStrategy(...) through transportExtension. Supply a configured Kafka client, consumer group, input topics, dead-letter topic, partition concurrency and error sink. The adapter owns only the producer and consumer it creates.

Kafka is event-only. Successful handling and confirmed dead-letter publication advance ordered offsets; retry pauses and rewinds only the affected partition. Attempt counts reset after restart or reassignment. A bounded close fences late settlement. The Kafka contract defines every option and lifecycle boundary.

Redis Pub/Sub#

const redis = createRedisStrategy({
  connection: { url: process.env.REDIS_URL },
  channels: ['orders.get'],
  channelPatterns: ['orders.events.*'],
  onError: error => transportErrors.report(error),
});

This is Redis Pub/Sub, not Streams. A message published while no matching subscriber is connected is lost. There is no acknowledgement, redelivery or dead-letter destination, so deliveryAttempt is always 1 and transportExtension({ transports: [redis], dispatcher }) requires dispatcher.onUndeliverable.

Exact and glob subscriptions dispatch the concrete channel. Request/reply uses a process-owned reply-channel prefix and still requires the caller's explicit deadline.

Core NATS#

const nats = createNatsStrategy({
  connection: { servers: process.env.NATS_URL },
  subscriptions: [{ subject: 'orders.*', queue: 'orders-workers' }, { subject: 'audit.>' }],
  onError: error => transportErrors.report(error),
});

This is core NATS, not JetStream. Delivery is at-most-once: there is no acknowledgement, redelivery or dead-letter destination. Native * and final > subscriptions are compiled into a trie at construction, and each delivery matches that trie rather than scanning the configured patterns. Queue groups are passed to NATS unchanged; the concrete delivered subject is the dispatcher pattern.

Core NATS also requires dispatcher.onUndeliverable. Request/reply uses an inbox subscription that is removed on reply, timeout, abort or publish failure.

RabbitMQ#

const rabbit = createRabbitMqStrategy({
  connection: env.RABBITMQ_URL,
  exchange: 'orders',
  queue: 'orders.worker',
  bindings: ['orders.*'],
  prefetch: 32,
  retry: {
    exchange: 'orders.retry',
    queue: 'orders.worker.retry',
  },
  deadLetter: {
    exchange: 'orders.dead',
    queue: 'orders.worker.dead',
  },
  onError: error => transportErrors.report(error),
});

prefetch is required and must be a positive integer; it is RabbitMQ's consumer backpressure control. The strategy declares topic exchanges, the main queue, a per-message-TTL retry queue and an owned dead-letter queue.

On retry, it publisher-confirm-publishes a copy to the retry queue with expiration: afterMs, then acknowledges the original. The retry queue dead-letters the expired copy back to the main exchange. On dead, it calls basic.nack(requeue: false) so the main queue routes the original to the configured DLQ. There is deliberately no nack(requeue: true): immediate head-of-queue requeue turns deterministic failures into a tight poison-message loop.

Broker delivery is never a substitute for an application transaction. Use the transactional outbox when a database write and event publication must not become a dual write.

Lifecycle and security#

Attach strategies through transportExtension in createApp({ extensions }). They open after application bootstrap, stop intake before dependencies are disposed, drain in-flight dispatch under graceMs and close in reverse declaration order.

Connection authentication, TLS and credential rotation are broker-client configuration. Do not trust identity claims carried in the payload, and do not put secrets in retained or dead-lettered messages. Handler effects must remain idempotent because RabbitMQ can redeliver an unacknowledged message.

Deferred transports#

Kafka is deferred because a consumer-group offset commits every earlier record in the partition; that cannot express independent settlement of concurrent messages through this interface. MQTT is deferred because broker QoS controls redelivery on the broker's schedule and cannot honour retry.afterMs.

No bespoke TCP framing is shipped.

---

See also: Microservices · Custom Transports · Hybrid Applications

SQS standard queues#

Install @zmdb/transport@1.0.0-beta.2, @zmdb/app@1.0.0-beta.2 and @aws-sdk/client-sqs@3.1127.0, then import @zmdb/transport/sqs. Supply a caller-owned SQSClient, distinct source and dead-letter queue URLs, and the explicit polling, visibility, concurrency and timeout options documented in the package SPEC. Attach createSqsStrategy(options) with transportExtension.

Successful dispatch deletes the current receipt. Retry changes visibility; dead-letter handling waits for the destination send before deleting the source. Closing stops polling and drains accepted work within its grace period; the caller destroys the client after app shutdown. This event transport advertises true / true / false capabilities and refuses FIFO and request/response. Standard queues can redeliver, so handlers must account for duplicates. Local acceptance uses the real AWS SDK with isolated ElasticMQ and WireMock fixtures.