Optional ReadonlybehaviorsIngress 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.
Optional ReadonlychainBounds 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.
OptionalclientOptionalconsumerDurable group retries; false retains nack-and-discard for operator DLX policies.
Default maxAttempts 5, delaysMs [5000, 30000, 120000, 600000].
Deterministic failures dead-letter immediately. Private queues and RPC reply inboxes discard.
With retries enabled, group suffixes .dead and .retry.
OptionaldeadQ.dead retention cap (positive safe integer, default 10000). Changing it requires draining and deleting the existing dead queue before restart.
OptionaldefaultOptionalexchangeOptional ReadonlyinboxThe 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.
OptionalnameInstance name for multi-instance support.
Optional ReadonlyoutboxThe 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.
OptionalpersistentPublish every message persistent (delivery_mode 2). Default true.
Consumer-group queues are declared durable, and a durable queue keeps
only PERSISTENT messages across a broker restart: before 0.9.0 every
message was published transient, so a RabbitMQ restart emptied every
queue of messages not yet consumed. false restores that behaviour, for
deliberately ephemeral traffic where losing in-flight messages on a
restart is acceptable in exchange for skipping the broker's disk write.
OptionalprefetchPer-consumer unacked delivery limit, integer 1–65535, default 32.
OptionalpublishBound on one publish(), in milliseconds. Default 15000; 0 waits
without a bound.
It covers every broker round trip a publish makes — the exchange assert,
and, on a confirm channel (a real amqplib connection always opens one),
the wait for RabbitMQ to accept the message. publish() resolves once
RabbitMQ has accepted it, rejects when it refuses it or the channel closes
first, and rejects when the bound expires: a paused broker keeps its
socket open, so without a bound the call would stay pending forever. A
rejection is not proof the message was dropped — the broker may still
accept it after the bound. A value outside 0–2147483647 (including
NaN) throws RangeError when MessagingPlugin(...) is called.
OptionalserializerSerializer for message payloads.
Optional ReadonlysubscriptionsSubscriptions 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.
OptionaltracingWhether to create producer and consumer spans when telemetry is available.
Optionalurl
RabbitMQ arm.
Since
0.1.0