Appearance

API reference@connectum/events-amqpDocs 1.3.x

Connectum API Reference / @connectum/events-amqp / types / AmqpAdapterOptions

Interface: AmqpAdapterOptions ​

Defined in: packages/events-amqp/src/types.ts:10

Options for creating an AMQP/RabbitMQ adapter.

Properties ​

consumerOptions? ​

readonly optional consumerOptions?: AmqpConsumerOptions

Defined in: packages/events-amqp/src/types.ts:76

Consumer options.

exchange? ​

readonly optional exchange?: string

Defined in: packages/events-amqp/src/types.ts:28

Exchange name for publishing and subscribing.

Default ​

ts
"connectum.events"

exchangeOptions? ​

readonly optional exchangeOptions?: AmqpExchangeOptions

Defined in: packages/events-amqp/src/types.ts:66

Exchange assertion options.

exchangeType? ​

readonly optional exchangeType?: "headers" | "topic" | "direct" | "fanout"

Defined in: packages/events-amqp/src/types.ts:61

Exchange type. The EventBus selects handlers by matching each delivered event type against the subscription patterns and acknowledges events without a handler; the adapter only guarantees that the broker delivers every wanted event, never a subset of them.

  • "topic": patterns are translated for the broker (* is one segment, a terminal > becomes *.#, one or more trailing segments). A complete # segment is rejected, since RabbitMQ would read it as a wildcard while the EventBus reads it as text.
  • "direct": the queue is bound to the exact routing key. With topologyMode: "assert" a subscription using a complete * or > is rejected, because the literal binding would never receive a message. With "check" or "skip" the adapter binds nothing, the operator's bindings decide delivery and the pattern only selects handlers.
  • "fanout": the routing key is ignored, so the queue receives every message whatever the pattern; the EventBus dispatches only matching handlers and acknowledges the rest, while a handler passed straight to adapter.subscribe() sees every message.
  • "headers": the adapter publishes no header carrying the event type, so it cannot derive header bindings from a pattern. In "assert" mode it binds the queue without arguments, which matches every message (the behavior of "fanout"), unless topology.bindings binds that queue to the exchange: then those declared bindings alone decide what the queue receives and the adapter adds none. A binding created outside the topology is not seen, and the argument-less binding would also deliver what it excludes; use "check" or "skip" for such bindings, or declare them in the topology.

Default ​

ts
"topic"

failFastOnInitialSetupError? ​

readonly optional failFastOnInitialSetupError?: boolean

Defined in: packages/events-amqp/src/types.ts:197

Fail fast on a topology setup failure on the FIRST connect, instead of entering amqplib's infinite recovery loop.

amqplib's opt-in recovery resolves connect() only after its setup hook succeeds, and rejects only once maxRetries is exhausted (default Infinity). A permanent topology error on the first connect under the default recovery therefore HANGS connect() forever, with no thrown error and — because the lifecycle listeners attach only after that never-returning await — no callback. When this flag is true (and recovery is enabled), a topology error of the initial connect rejects connect() with the typed AmqpTopologyError / AmqpConnectionError. Without AmqpRecoveryOptions.initialConnectMaxRetries the adapter first validates topology against a throwaway non-recovering connection (the startup probe); with it, every initial attempt runs the full setup and the first setup error stops the initial connect.

Only a failure of the topology setup pass reported as AmqpTopologyError fails fast — a broker reply to a declare/check/bind (404, 406, 403, …) or the adapter's own configuration check. A broker that cannot be reached, and a connection lost while the pass runs (reported as AmqpConnectionError), are not fail-fast: the adapter continues with recovery (block-until-broker, or the initial connect budget). SUBSEQUENT reconnects always keep infinite-recovery behavior.

No-op with recovery: false (that path already fails fast on setup). Without initialConnectMaxRetries, enabling this — or supplying AmqpLifecycleCallbacks.onLifecycle or AmqpLifecycleCallbacks.onSetupFailed — adds one extra short-lived connection plus a topology validation pass at startup for the probe (recovery must be enabled; with recovery: false no probe runs and no setup-failed event is delivered).

Default ​

ts
false

lifecycle? ​

readonly optional lifecycle?: AmqpLifecycleCallbacks

Defined in: packages/events-amqp/src/types.ts:323

Connection lifecycle callbacks. Connection errors are surfaced here — not just logged.

publisherOptions? ​

readonly optional publisherOptions?: AmqpPublisherOptions

Defined in: packages/events-amqp/src/types.ts:81

Publisher options.

publishRetry? ​

readonly optional publishRetry?: boolean | AmqpPublishRetryOptions

Defined in: packages/events-amqp/src/types.ts:317

Opt-in bounded publish retry for CONNECTION-CLASS outcomes (since 1.3.0).

When enabled, a publish() that fails with AmqpConnectionError — publishing during a recovery window, or an in-flight confirm lost to a connection drop — is retried in place (the caller's promise stays pending) instead of rejecting immediately: a short broker blip becomes a transparent delay. true = defaults; an object tunes the budget.

The retry boundary is isAutoRetriablePublishError — deliberately NARROWER than the at-least-once republish matrix in the error taxonomy: a broker nack is republish-safe by policy but is NOT auto-retried inline (it is an explicit broker refusal, e.g. an over-capacity queue — hammering it in a tight loop helps nobody). AmqpPublishTimeoutError joins the boundary only with retryOnTimeout: true.

Semantics — read before enabling:

  • At-least-once, full stop. A retried publish whose previous attempt was lost IN FLIGHT (confirm never arrived: state UNKNOWN) may duplicate on the broker. x-event-id / messageId stay STABLE across attempts (also with externalContract — a caller-supplied id is reused as-is), so consumer-side dedup keys on them.

  • Worst-case latency: each attempt is bounded by publishTimeoutMs (default 30s), so maxRetries: 5 can hold a single publish() for several minutes worst-case — far beyond typical 30s RPC timeouts. There is deliberately no second overall-deadline knob: bound the budget via maxRetries/publishTimeoutMs.

  • Shutdown-aware: the loop aborts on disconnect() (throws the last connection error) and, living inside the adapter.publish() promise, is automatically covered by the bus-level drainPublishTimeout.

  • Single-flight (mandatory: true with correlationHeader: false; externalContract forces the latter but single-flight still requires mandatory): retries hold the chain — ordering is preserved at the cost of head-of-line blocking during backoff. In this headerless mode a late basic.return from an abandoned timed-out attempt may mark the current one (correlation is attempt-agnostic without the header) — prefer the default header correlation when combining mandatory with retryOnTimeout.

  • A broker-closed publish channel is not retried: when the broker closes the publish CHANNEL with a reply code of any kind — 404 (a publish to a missing exchange under topologyMode: "skip"), 403 (a publish to an internal exchange), 406, 541, … — and that channel is still the current one, the publish surfaces immediately with the broker reply as cause, without onRetry or backoff. The connection stays up and recovery only recreates the channel when the connection itself drops, so retrying cannot heal. If recovery has already replaced the channel, retrying continues normally.

  • maxRetries: Infinity makes publish() wait until recovery heals the connection or disconnect() is called. A channel-only close with a reply code is not part of that wait (see the previous item).

  • No retry against a dead cycle: once recovery has given up (terminal reconnect-failed) or recovery: false lost its connection, nothing can heal the publish, so it rejects at once without spending the budget.

Backoff mirrors the recovery formula (same knob names and semantics; a delay never exceeds maxDelay), but the DEFAULT budget differs: maxRetries here defaults to 5 (bounded), not Infinity.

Default ​

ts
undefined (disabled — behavior unchanged)

publishTimeoutMs? ​

readonly optional publishTimeoutMs?: number

Defined in: packages/events-amqp/src/types.ts:338

Per-publish broker-outcome deadline in milliseconds. A publish whose ack/nack/return/connection-loss outcome does not arrive in time rejects with AmqpPublishTimeoutError (message state UNKNOWN — an at-least-once producer should republish).

A value that is not a finite number of at least 1 (NaN, Infinity, 0, a negative number) counts as unset and the default applies; a fraction is floored; a value above 2147483647 (the largest delay a timer honors) is capped to it. There is no "no timeout" value.

Default ​

ts
30000

queueOptions? ​

readonly optional queueOptions?: AmqpQueueOptions

Defined in: packages/events-amqp/src/types.ts:71

Default queue assertion options.

queueOverrides? ​

readonly optional queueOverrides?: Record<string, AmqpQueueOverride>

Defined in: packages/events-amqp/src/types.ts:127

Map a consumer group name to an externally-named queue.

By default a group consumes from ${exchange}.${group}. An override lets a subscription attach to a queue from an external contract (with its own arguments) instead.

recovery? ​

readonly optional recovery?: boolean | AmqpRecoveryOptions

Defined in: packages/events-amqp/src/types.ts:161

Automatic connection recovery (delegated to amqplib's opt-in recovery). Enabled by default; pass false to restore no-reconnect behavior.

On every (re)connect the adapter re-creates its channels, re-applies topology (per topologyMode), and replays active subscriptions. In-flight publishes at the moment of a connection loss reject with AmqpConnectionError.

With recovery enabled the adapter also restores a consumer the broker ended while the connection stayed up (queue deleted, consumer cancelled, channel closed): consumer-lost, then consumer-restored or consumer-restore-failed. Attempts wait the same delay formula (the numeric knobs; the defaults when backoff is set) and have no limit except a failure that cannot heal. With false the loss is reported with willRestore: false and the consumer stays dead.

maxRetries governs BOTH the initial connect and steady-state recovery (counter reset on success); under the default Infinity, connect() blocks until the broker is reachable rather than failing fast (see AmqpAdapterOptions.failFastOnInitialSetupError to fail fast on a deterministic startup misconfiguration). See AmqpRecoveryOptions for the retry-budget scope and the delay bounds. With a finite maxRetries, a broker still unreachable when the initial connect has used its budget rejects connect() with a typed AmqpConnectionError whose cause is the last connection error. Once a finite budget is exhausted, recovery is over for good — see the terminal reconnect-failed in AmqpLifecycleEvent.

Default ​

ts
true (amqplib defaults: 100ms initial, ×2, 30s cap, jitter 0.2, infinite retries)

serialization? ​

readonly optional serialization?: AmqpSerializationOptions

Defined in: packages/events-amqp/src/types.ts:92

Message serialization metadata and optional wire transcoding.

The adapter receives payloads as bytes (the EventBus serializes protobuf upstream); this option controls the AMQP contentType property and lets an application transcode the wire body — e.g. when the application serializes JSON itself and publishes through the adapter directly against an external AsyncAPI contract.

socketOptions? ​

readonly optional socketOptions?: Record<string, unknown>

Defined in: packages/events-amqp/src/types.ts:21

Socket options passed to amqplib.connect().

topology? ​

readonly optional topology?: AmqpTopology

Defined in: packages/events-amqp/src/types.ts:100

Explicit topology to declare on connect (and re-declare after recovery): exchanges, queues with arbitrary external names and raw arguments (e.g. x-dead-letter-exchange), and bindings — including exchange-to-exchange.

topologyMode? ​

readonly optional topologyMode?: AmqpTopologyMode

Defined in: packages/events-amqp/src/types.ts:118

How topology is established:

  • "assert" (default) — declare idempotently (assertExchange/assertQueue/bind);
  • "check" — existence-only verification (checkExchange/checkQueue). A missing object raises AmqpTopologyError, which fails connect() fast ONLY with recovery: false or failFastOnInitialSetupError: true; under the default recovery a first-connect check failure otherwise enters the (infinite) recovery loop and is surfaced via onSetupFailed / onReconnecting rather than rejecting connect(). AMQP offers no passive introspection: argument equivalence and binding presence are NOT verifiable in this mode (a conflicting redeclare elsewhere is PRECONDITION_FAILED 406);
  • "skip" — no topology operations at all; the application owns topology.

Default ​

ts
"assert"

treatTopologyErrorAsFatal? ​

readonly optional treatTopologyErrorAsFatal?: boolean

Defined in: packages/events-amqp/src/types.ts:253

Treat DETERMINISTIC topology drift during steady-state recovery as fatal: stop the reconnect cycle instead of retrying forever against a misconfigured broker.

Under the default maxRetries: Infinity, a queue/exchange deleted or redeclared incompatibly while the adapter is reconnecting makes every recovery attempt fail deterministically — the adapter would retry forever, reporting setup-failed on each attempt but never giving up. With this flag the adapter stops the cycle on the first such failure and reports the terminal reconnect-failed lifecycle event (after the setup-failed event for the same attempt); subsequent publishes fail fast with AmqpConnectionError.

The gate is the broker reply of the failure cause (its AMQP reply code AND reply text) — NOT the error class: transient causes wrapped into AmqpTopologyError during a setup pass (broker restarting 320, internal error 541, resource locked 405, a mid-setup connection drop) stay in normal recovery. Only replies that name a condition which cannot heal without a configuration or topology change are fatal:

  • 404 "no queue" / "no exchange" (the object is missing);
  • 406 "inequivalent arg" (redeclare with different arguments), "invalid arg", "unknown exchange type" / "invalid exchange type".

Everything else stays in recovery, including a 404 that RabbitMQ raises for a self-healing queue condition (home node down or inaccessible, queue process crashed or stopped by its supervisor, timeout, leader stopping or being demoted), a 406 such as "exchange limit reached" (clears when exchanges are removed), and a reply that carries no text. A RabbitMQ release that rewords a message therefore degrades to "keep retrying" (visible through reconnecting / setup-failed), never to a silent permanent stop.

A fatal stop is quiet: no exception reaches application code. It is observable only through lifecycle.onLifecycle (reconnect-failed) or lifecycle.onReconnectFailed, and after it the adapter stays down until the application starts the bus again — wire one of those callbacks and restart from there.

After the fatal stop the adapter is fully torn down: consumers are dead (subscription records are cleared, mirroring disconnect()), publishes fail fast, and a later connect() starts from a clean slate — re-subscribe explicitly.

Scope: steady-state recovery only. Boot-time drift is the startup probe's job — see failFastOnInitialSetupError. Setting both covers boot and steady state; the remaining window — broker unreachable at connect() time with drift surfacing before the first successful connect — is closed by AmqpRecoveryOptions.initialConnectMaxRetries (since 1.3.0), which reports those failures per attempt and rejects on exhaustion.

Default ​

ts
false

url ​

readonly url: string

Defined in: packages/events-amqp/src/types.ts:16

AMQP connection URL.

Example ​

ts
"amqp://guest:guest@localhost:5672"