Interface RedisStreamsMessagingOptions

Ingress behaviours wrapped around every subscription handler — the messaging arm of the transport-neutral behaviour chain shared with the websocket, queue, and scheduler plugins (IIngressBehavior in @setu-ts/common).

Each behaviour observes an IngressContext carrying kind: 'messaging', the topic as name, the delivered message as payload, and the transport headers from MessageMetadata (absent when the transport carried no channel — there is deliberately NO attempt: brokers redeliver, but MessageMetadata exposes no delivery count), and runs in declared order ahead of the handler. A behaviour that returns without calling next() short-circuits: the handler never sees the message. A behaviour that throws follows the messaging handler's existing rejection path. The chain wraps SUBSCRIBE handlers only — respond (RPC) is deliberately not chained and not armed in this milestone.

With no behaviours configured, the broker chain is byte-identical to the pre-arm behaviour: no PipelinedBroker decorator is applied at all.

Instance entries are read by the chain at register(); factory entries are resolved in the onInit phase and a throwing factory rejects start() naming MessagingPlugin({ behaviors }) and the entry's index in THIS declared array.

When an entry is a FACTORY, DELIVERY is held until onInit has resolved the whole chain, so no message reaches a handler through a partial one. A broker holding a backlog delivers the moment a consumer attaches, and the gate covers every subscription — this plugin's declared entries and any a later plugin makes imperatively through the resolved broker — which is why no registration's timing has to change. It is released once and costs nothing thereafter.

Bounds a dispatch held on the behaviour-chain gate, which exists only when a RegistryFactory behaviour is declared. A held dispatch that waits longer than this rejects with ChainGateTimeoutError, whose message names the likely cause (a plugin publishing during its own register()); the gate itself is left in place, so later dispatches refuse the same way rather than delivering through a partial chain.

Default 10_000 ms. 0 disables the bound and restores the wait-forever behaviour, for an application that would rather hang than fail. The value must be a finite, non-negative number no greater than 2_147_483_647: NaN, a negative value, Infinity, or a larger value throws a RangeError naming this option at registration, when a behaviour factory arms the gate (Infinity does NOT mean wait-forever — 0 does). Ignored entirely when no behaviour factory is configured, because the gate is then never armed. Declared on this shared base — never per-arm — since the gate exists for every broker.

The consumer inbox (M108). When set, the plugin registers an IInbox under CAPABILITIES.INBOX (inbox.<name> for a named instance) and an inbox health indicator, so that onIntegrationEvent(definition, handler, { inbox: { consumer } }) subscriptions apply each event's database writes once per consumer. At onInit it resolves the store, runs its verify() — a refusal fails start() — and, unless purge.schedule is false, schedules the retention purge (inbox-purge[.<name>]) on CAPABILITIES.SCHEDULER; with no scheduler registered start() rejects InboxPurgeUnscheduledError. The inbox closes after the broker disconnects, so no in-flight delivery is refused during a stop.

Every numeric option is validated when MessagingPlugin(...) is called. Absent, nothing changes: no capability, hook, indicator or ordering edge is added.

Instance name for multi-instance support.

The transactional outbox (M107). When set, the plugin registers an IOutbox under CAPABILITIES.OUTBOX (outbox.<name> for a named instance) and an outbox health indicator. At onInit it resolves the store(s), runs each store's verify() — a refusal fails start() — and, unless relay.schedule is false, schedules the relay (outbox-relay[.<name>], every relay.intervalMs) and the purge (outbox-purge[.<name>], every purgeIntervalMs) on CAPABILITIES.SCHEDULER; with no scheduler registered start() rejects OutboxRelayUnscheduledError. On shutdown the relay is drained in an onShutdown hook, before any close hook — the broker's included.

Every numeric option is validated when MessagingPlugin(...) is called. Absent, nothing changes: no capability, hook, indicator or ordering edge is added.

Serializer for message payloads.

Subscriptions registered declaratively, as an alternative to calling broker.subscribe(topic, handler, options) imperatively after start(). Each entry — instance or RegistryFactory — produces one subscribe() call, so a subscription can be declared where the plugin is composed instead of after the application has started.

Instance entries register during the plugin's register() phase, identical to the imperative timing. Factory entries are resolved in the onInit phase — the first at which the registry holds every capability — and the plugin AWAITS each subscribe() there, so the subscription is established before the application serves. A factory that throws rejects start() with an error naming MessagingPlugin({ subscriptions }) and the entry's index in THIS declared array, not its position among the factories.

Because the arm sits on this shared interface, EVERY MessagingPluginOptions union arm inherits it — the declarative form is broker-agnostic, exactly like the imperative subscribe() it mirrors.

Whether to create producer and consumer spans when telemetry is available.