@amqp-contract/core

@amqp-contract/core

Classes

AmqpClient

Defined in: packages/core/src/amqp-client.ts:311

AMQP client that manages connections and channels with automatic topology setup.

This class handles:

Connection management with automatic reconnection via amqp-connection-manager
Connection pooling and sharing across instances with the same URLs
Automatic AMQP topology setup (exchanges, queues, bindings) from contract
Content encoding: non-Buffer payloads are JSON-encoded at publish time, Buffers go on the wire byte-for-byte

Two failures are modeled: an unreachable broker (waitForConnect → ConnectionError) and a broker-side publish failure (publish / sendToQueue → PublishError). Every other infrastructure failure is unexpected, so it surfaces through the Defect channel (with a TechnicalError as the defect's cause for logging).

Example

typescript

const client = new AmqpClient(contract, {
  urls: ['amqp://localhost'],
  connectionOptions: { heartbeatIntervalInSeconds: 30 }
});

// Wait for connection (AsyncResult is thenable)
await client.waitForConnect();

// Publish a message
const result = await client.publish(
  { exchange: 'exchange', routingKey: 'routingKey' },
  { data: 'value' },
);

// Close when done
await client.close().get();

Constructors

Constructor

new AmqpClient(contract, options): AmqpClient;

Defined in: packages/core/src/amqp-client.ts:384

Create a new AMQP client instance.

The client will automatically:

Get or create a shared connection using the singleton pattern
Set up AMQP topology (exchanges, queues, bindings) from the contract
Create a confirm channel that encodes content at publish time (JSON for plain values, byte-for-byte for Buffers)

Parameters

Parameter	Type	Description
`contract`	`ContractDefinition`	The contract definition specifying the AMQP topology
`options`	`AmqpClientOptions`	Client configuration options

Returns

AmqpClient

Accessors

currentChannelEpoch

Get Signature

get currentChannelEpoch(): number;

Defined in: packages/core/src/amqp-client.ts:751

The current channel epoch — bumped on every channel 'connect' (initial connect included). Consumers stamp deliveries with this value and pass it back via ack / nack so a settle can never target a tag from a previous channel incarnation.

Returns

number

Methods

ack()

ack(msg, options?): void;

Defined in: packages/core/src/amqp-client.ts:790

Acknowledge a message.

Parameters

Parameter	Type	Description
`msg`	`ConsumeMessage`	The message to acknowledge
`options?`	{ `allUpTo?`: `boolean`; `deliveryEpoch?`: `number`; }	Settle options: - `allUpTo` — if true, acknowledge all messages up to and including this one (defaults to false). - `deliveryEpoch` — pass the epoch captured when the message was delivered (currentChannelEpoch) to make the ack reconnect-safe: a stale epoch skips the ack (logged) instead of settling a foreign tag.
`options.allUpTo?`	`boolean`	-
`options.deliveryEpoch?`	`number`	-

Returns

void

addSetup()

addSetup(setup): void;

Defined in: packages/core/src/amqp-client.ts:835

Add a setup function to be called when the channel is created or reconnected.

The only way to register channel-level setup after construction — channelOptions.setup is fixed at construction time — so it is what a caller holding an already-built client reaches for.

Parameters

Parameter	Type	Description
`setup`	(`channel`) => `void` \| `Promise`<`void`>	The setup function to add

Returns

void

cancel()

cancel(consumerTag): AsyncResult<void, never>;

Defined in: packages/core/src/amqp-client.ts:739

Cancel a consumer by its consumer tag.

Parameters

Parameter	Type
`consumerTag`	`string`

Returns

AsyncResult<void, never>

close()

close(): AsyncResult<void, never>;

Defined in: packages/core/src/amqp-client.ts:869

Close the channel and release the connection lease.

This will:

Close the channel wrapper
Release this client's lease on the shared connection
Close the connection if this was the last client using it

Idempotent: a second close() returns the same in-flight (or settled) result instead of double-releasing the shared connection.

Both steps run regardless of each other's outcome; if both fail, the errors are wrapped in an AggregateError.

Returns

AsyncResult<void, never>

consume()

consume(
   queue, 
   callback, 
   options?
): AsyncResult<string, never>;

Defined in: packages/core/src/amqp-client.ts:695

Start consuming messages from a queue.

options.prefetch maps to amqp-connection-manager's native per-consumer prefetch: applied via basic.qos(count, global=false) immediately before this consumer's basic.consume, and re-applied the same way when the consumer is re-established after a reconnect — so the value never bleeds onto other consumers sharing the channel.

Parameters

Parameter	Type
`queue`	`string`
`callback`	`ConsumeCallback`
`options?`	`AmqpConsumeOptions`

Returns

AsyncResult<string, never>

AsyncResult resolving to the consumer tag.

getConnection()

getConnection(): IAmqpConnectionManager;

Defined in: packages/core/src/amqp-client.ts:506

The underlying connection manager.

The connection is shared across every AmqpClient created with the same URLs and connection options, so two clients that pool together return the identical instance — which is how the connection-sharing suite asserts pooling.

Returns

IAmqpConnectionManager

isConnected()

isConnected(): boolean;

Defined in: packages/core/src/amqp-client.ts:515

Whether the broker connection is up right now and this client has not been closed — the answer a readiness probe wants. false while amqp-connection-manager is reconnecting.

Returns

boolean

nack()

nack(msg, options?): void;

Defined in: packages/core/src/amqp-client.ts:811

Negative acknowledge a message.

Parameters

Parameter	Type	Description
`msg`	`ConsumeMessage`	The message to nack
`options?`	{ `allUpTo?`: `boolean`; `deliveryEpoch?`: `number`; `requeue?`: `boolean`; }	Settle options: - `allUpTo` — if true, nack all messages up to and including this one (defaults to false). - `requeue` — if true, requeue the message(s) (defaults to true). - `deliveryEpoch` — pass the epoch captured at delivery time to make the nack reconnect-safe (see ack).
`options.allUpTo?`	`boolean`	-
`options.deliveryEpoch?`	`number`	-
`options.requeue?`	`boolean`	-

Returns

void

on()

on(event, listener): void;

Defined in: packages/core/src/amqp-client.ts:850

Available events:

'connect': Emitted when the channel is (re)connected
'close': Emitted when the channel is closed
'error': Emitted when an error occurs

Parameters

Parameter	Type	Description
`event`	`string`	The event name
`listener`	(...`args`) => `void`	The event listener

Returns

void

publish()

publish(
   target, 
   content, 
   options?
): AsyncResult<void, PublishError>;

Defined in: packages/core/src/amqp-client.ts:609

Publish a message to an exchange.

Non-Buffer content is JSON-encoded; Buffers are published byte-for-byte.

A broker-side failure core can name — publish timeout, broker nack, channel closed — is the modeled PublishError. A confirmed publish that leaves the write buffer full is a success (logged at debug). An unencodable payload or an unrecognised rejection is a Defect with a TechnicalError cause.

Parameters

Parameter	Type	Description
`target`	{ `exchange`: `string`; `routingKey`: `string`; }	The exchange and routing key to publish to
`target.exchange`	`string`	-
`target.routingKey`	`string`	-
`content`	`unknown`	The message payload
`options?`	`Publish`	AMQP publish options

Returns

AsyncResult<void, PublishError>

sendToQueue()

sendToQueue(
   queue, 
   content, 
   options?
): AsyncResult<void, PublishError>;

Defined in: packages/core/src/amqp-client.ts:627

Publish a message directly to a queue.

Non-Buffer content is JSON-encoded; Buffers are published byte-for-byte. Failures are reported exactly as for publish.

Parameters

Parameter	Type
`queue`	`string`
`content`	`unknown`
`options?`	`Publish`

Returns

AsyncResult<void, PublishError>

waitForConnect()

waitForConnect(): AsyncResult<void, ConnectionError>;

Defined in: packages/core/src/amqp-client.ts:536

Wait for the channel to be connected and ready.

The returned AsyncResult resolves to Err({@link ConnectionError}) once connectTimeoutMs elapses — modeled, because a broker that will not answer is an operator's business rather than a bug. That timeout defaults to DEFAULT_CONNECT_TIMEOUT_MS when the option is omitted; only an explicit null disables it, and then this waits forever — amqp-connection-manager retries connections indefinitely and never errors on its own.

NOTE: When using AmqpClient directly (not via TypedAmqpClient / TypedAmqpWorker), the constructor has already incremented the pooled connection's reference count. Callers must invoke close() on the failure path to release the connection — waitForConnect does not do this automatically. The typed factories handle this cleanup for you.

Returns

AsyncResult<void, ConnectionError>

ConnectionError

Defined in: packages/core/src/errors.ts:71

The broker could not be reached: refused, unresolvable, unauthorized, or still not ready when connectTimeoutMs elapsed.

Modeled, not a defect — unlike TechnicalError. An unreachable broker is the anticipated failure of dialing one: it is what a wrong URL, a rotated credential or a cluster that has not come up yet look like, every one of them an operator's business rather than a bug in the caller. So TypedAmqpWorker.create and TypedAmqpClient.create report it on the E channel, where a start-up path can triage it by tag and turn it into an exit code, a retry, or a health probe — and the defect channel keeps its meaning: the failures nobody anticipated.

Carries a _tag of "@amqp-contract/ConnectionError"; the human-facing Error.name is kept bare ("ConnectionError"). The underlying amqplib rejection is on cause.

Extends

TaggedErrorInstance<"@amqp-contract/ConnectionError", { cause?: unknown; }>

Constructors

Constructor

new ConnectionError(message, cause?): ConnectionError;

Defined in: packages/core/src/errors.ts:79

Parameters

Parameter	Type
`message`	`string`
`cause?`	`unknown`

Returns

ConnectionError

Overrides

TaggedError("@amqp-contract/ConnectionError", { name: "ConnectionError", })<{ cause?: unknown; }>.constructor

Properties

Property	Modifier	Type	Default value	Description	Inherited from	Defined in
`_tag`	`readonly`	`"@amqp-contract/ConnectionError"`	`undefined`	-	`TaggedError("@amqp-contract/ConnectionError", { name: "ConnectionError", })._tag`	node_modules/.pnpm/unthrown@5.12.2/node_modules/unthrown/dist/index.d.mts:2363
`cause?`	`public`	`unknown`	`undefined`	-	`MessageValidationError`.`cause`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es2022.error.d.ts:24
`message`	`public`	`string`	`undefined`	-	`TaggedError("@amqp-contract/ConnectionError", { name: "ConnectionError", }).message`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es5.d.ts:1075
`name`	`public`	`string`	`undefined`	-	`TaggedError("@amqp-contract/ConnectionError", { name: "ConnectionError", }).name`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es5.d.ts:1074
`stack?`	`public`	`string`	`undefined`	-	`TaggedError("@amqp-contract/ConnectionError", { name: "ConnectionError", }).stack`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es5.d.ts:1076
`tag`	`static readonly`	`"@amqp-contract/ConnectionError"`	`"@amqp-contract/ConnectionError"`	The `_tag`, for `P.tag(ConnectionError.tag)` without a raw string.	-	packages/core/src/errors.ts:77

MessageValidationError

Defined in: packages/core/src/errors.ts:151

Error thrown when message validation fails (payload or headers).

Used by both the client (publish-time payload validation) and the worker (consume-time payload and headers validation). Carries a _tag of "@amqp-contract/MessageValidationError" (namespaced to avoid collisions); the Error.name is kept bare ("MessageValidationError").

Param

source

The name of the publisher or consumer that triggered the validation

Param

issues

The validation issues from the Standard Schema validation

Extends

TaggedErrorInstance<"@amqp-contract/MessageValidationError", { issues: unknown; source: string; }>

Constructors

Constructor

new MessageValidationError(source, issues): MessageValidationError;

Defined in: packages/core/src/errors.ts:160

Parameters

Parameter	Type
`source`	`string`
`issues`	`unknown`

Returns

MessageValidationError

Overrides

TaggedError("@amqp-contract/MessageValidationError", { name: "MessageValidationError", })<{ source: string; issues: unknown; }>.constructor

Properties

Property	Modifier	Type	Default value	Description	Inherited from	Defined in
`_tag`	`readonly`	`"@amqp-contract/MessageValidationError"`	`undefined`	-	`TaggedError("@amqp-contract/MessageValidationError", { name: "MessageValidationError", })._tag`	node_modules/.pnpm/unthrown@5.12.2/node_modules/unthrown/dist/index.d.mts:2363
`cause?`	`public`	`unknown`	`undefined`	-	`TaggedError("@amqp-contract/MessageValidationError", { name: "MessageValidationError", }).cause`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es2022.error.d.ts:24
`issues`	`readonly`	`unknown`	`undefined`	-	`TaggedError("@amqp-contract/MessageValidationError", { name: "MessageValidationError", }).issues`	packages/core/src/errors.ts:155
`message`	`public`	`string`	`undefined`	-	`TaggedError("@amqp-contract/MessageValidationError", { name: "MessageValidationError", }).message`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es5.d.ts:1075
`name`	`public`	`string`	`undefined`	-	`TaggedError("@amqp-contract/MessageValidationError", { name: "MessageValidationError", }).name`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es5.d.ts:1074
`source`	`readonly`	`string`	`undefined`	-	`TaggedError("@amqp-contract/MessageValidationError", { name: "MessageValidationError", }).source`	packages/core/src/errors.ts:154
`stack?`	`public`	`string`	`undefined`	-	`TaggedError("@amqp-contract/MessageValidationError", { name: "MessageValidationError", }).stack`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es5.d.ts:1076
`tag`	`static readonly`	`"@amqp-contract/MessageValidationError"`	`"@amqp-contract/MessageValidationError"`	The `_tag`, for `P.tag(MessageValidationError.tag)` without a raw string.	-	packages/core/src/errors.ts:158

PublishError

Defined in: packages/core/src/errors.ts:122

The broker side of a publish failed: timed out, nacked, or the channel closed under it. (A full write buffer is NOT one: on the confirm channel the wrapper only reports it after the broker confirmed the message.)

Modeled, not a defect — a broker that is down, overloaded or refusing a message is an operational condition a publisher is expected to handle (buffer, retry, shed load, surface a 503), not a bug. Returned on the E channel of AmqpClient.publish, TypedAmqpClient.publish and TypedAmqpClient.call; switch on PublishError.reason. A failure core cannot classify (an unencodable payload, an unknown rejection) stays a Defect with a TechnicalError cause.

Carries a _tag of "@amqp-contract/PublishError"; the Error.name is kept bare ("PublishError"). The underlying rejection, if any, is on cause.

Extends

TaggedErrorInstance<"@amqp-contract/PublishError", { cause?: unknown; reason: PublishFailureReason; target: string; }>

Constructors

Constructor

new PublishError(props): PublishError;

Defined in: packages/core/src/errors.ts:133

Parameters

Parameter	Type
`props`	{ `cause?`: `unknown`; `reason`: `PublishFailureReason`; `target`: `string`; }
`props.cause?`	`unknown`
`props.reason`	`PublishFailureReason`
`props.target`	`string`

Returns

PublishError

Overrides

TaggedError("@amqp-contract/PublishError", { name: "PublishError", })<{ reason: PublishFailureReason; /** Where the message was going, e.g. `exchange "orders" (routing key "order.created")`. */ target: string; cause?: unknown; }>.constructor

Properties

Property	Modifier	Type	Default value	Description	Inherited from	Defined in
`_tag`	`readonly`	`"@amqp-contract/PublishError"`	`undefined`	-	`TaggedError("@amqp-contract/PublishError", { name: "PublishError", })._tag`	node_modules/.pnpm/unthrown@5.12.2/node_modules/unthrown/dist/index.d.mts:2363
`cause?`	`public`	`unknown`	`undefined`	-	`MessageValidationError`.`cause`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es2022.error.d.ts:24
`message`	`public`	`string`	`undefined`	-	`TaggedError("@amqp-contract/PublishError", { name: "PublishError", }).message`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es5.d.ts:1075
`name`	`public`	`string`	`undefined`	-	`TaggedError("@amqp-contract/PublishError", { name: "PublishError", }).name`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es5.d.ts:1074
`reason`	`readonly`	`PublishFailureReason`	`undefined`	-	`TaggedError("@amqp-contract/PublishError", { name: "PublishError", }).reason`	packages/core/src/errors.ts:125
`stack?`	`public`	`string`	`undefined`	-	`TaggedError("@amqp-contract/PublishError", { name: "PublishError", }).stack`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es5.d.ts:1076
`target`	`readonly`	`string`	`undefined`	Where the message was going, e.g. `exchange "orders" (routing key "order.created")`.	`TaggedError("@amqp-contract/PublishError", { name: "PublishError", }).target`	packages/core/src/errors.ts:127
`tag`	`static readonly`	`"@amqp-contract/PublishError"`	`"@amqp-contract/PublishError"`	The `_tag`, for `P.tag(PublishError.tag)` without a raw string.	-	packages/core/src/errors.ts:131

RpcError

Defined in: packages/core/src/errors.ts:233

A typed, contract-declared RPC error — the business-failure channel of an RPC, as opposed to transport failures (which surface as a Defect with a TechnicalError cause).

Declared per-RPC via defineRpc(queue, { request, response, errors }), where each error code maps to a message definition validating the error's data payload. A worker handler surfaces one by returning Err(rpcError(code, data)); the worker validates data against the declared schema, publishes an error reply, and acks the request (business errors are not retried). The caller's client.call(...) resolves to Err(RpcError<code, data>) with data re-validated on arrival.

Carries a _tag of "@amqp-contract/RpcError" for exhaustive dispatch via the error matcher (result.match({ ok, defect, errCases: (matcher) => matcher.with(P.tag("@amqp-contract/RpcError"), …) })); the Error.name is kept bare ("RpcError"). Discriminate between codes on the code property.

Extends

TaggedErrorInstance<"@amqp-contract/RpcError", { code: string; data: unknown; }>

Type Parameters

Type Parameter	Default type
`TCode` extends `string`	`string`
`TData`	`unknown`

Constructors

Constructor

new RpcError<TCode, TData>(
   code, 
   data, 
   message?
): RpcError<TCode, TData>;

Defined in: packages/core/src/errors.ts:246

Parameters

Parameter	Type
`code`	`TCode`
`data`	`TData`
`message?`	`string`

Returns

RpcError<TCode, TData>

Overrides

TaggedError( "@amqp-contract/RpcError", { name: "RpcError" }, )<{ code: string; data: unknown; }>.constructor

Properties

Property	Modifier	Type	Default value	Description	Overrides	Inherited from	Defined in
`_tag`	`readonly`	`"@amqp-contract/RpcError"`	`undefined`	-	-	`TaggedError( "@amqp-contract/RpcError", { name: "RpcError" }, )._tag`	node_modules/.pnpm/unthrown@5.12.2/node_modules/unthrown/dist/index.d.mts:2363
`cause?`	`public`	`unknown`	`undefined`	-	-	`TaggedError( "@amqp-contract/RpcError", { name: "RpcError" }, ).cause`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es2022.error.d.ts:24
`code`	`readonly`	`TCode`	`undefined`	-	`TaggedError( "@amqp-contract/RpcError", { name: "RpcError" }, ).code`	-	packages/core/src/errors.ts:243
`data`	`readonly`	`TData`	`undefined`	-	`TaggedError( "@amqp-contract/RpcError", { name: "RpcError" }, ).data`	-	packages/core/src/errors.ts:244
`message`	`public`	`string`	`undefined`	-	-	`TaggedError( "@amqp-contract/RpcError", { name: "RpcError" }, ).message`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es5.d.ts:1075
`name`	`public`	`string`	`undefined`	-	-	`TaggedError( "@amqp-contract/RpcError", { name: "RpcError" }, ).name`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es5.d.ts:1074
`stack?`	`public`	`string`	`undefined`	-	-	`TaggedError( "@amqp-contract/RpcError", { name: "RpcError" }, ).stack`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es5.d.ts:1076
`tag`	`static readonly`	`"@amqp-contract/RpcError"`	`"@amqp-contract/RpcError"`	The `_tag`, for `P.tag(RpcError.tag)` without a raw string.	-	-	packages/core/src/errors.ts:241

TechnicalError

Defined in: packages/core/src/errors.ts:39

Error for technical/runtime failures that cannot be prevented by TypeScript.

This includes channel issues, compression/parse faults, and other unexpected runtime errors. Shared across core, worker, and client packages. Dialing the broker is NOT one of them any more — that failure is anticipated, and it is modeled as ConnectionError.

These failures are unexpected, so @amqp-contract surfaces them through unthrown's defect channel, not the modeled E channel: a TechnicalError instance is carried as the cause of a Defect (so its message/cause survive for logging), and is handled in the defect arm of result.match({ ok, errCases, defect }) — or via recoverDefect / tapDefect — never matched in errCases. It is deliberately absent from every operation's E (only anticipated domain failures live there).

Built on unthrown's TaggedError, so it carries a _tag of "@amqp-contract/TechnicalError" (namespaced to avoid colliding with other libraries' tags); the human-facing Error.name is kept bare ("TechnicalError"). Remains a real Error.

Extends

TaggedErrorInstance<"@amqp-contract/TechnicalError", { cause?: unknown; }>

Constructors

Constructor

new TechnicalError(message, cause?): TechnicalError;

Defined in: packages/core/src/errors.ts:47

Parameters

Parameter	Type
`message`	`string`
`cause?`	`unknown`

Returns

TechnicalError

Overrides

TaggedError("@amqp-contract/TechnicalError", { name: "TechnicalError", })<{ cause?: unknown; }>.constructor

Properties

Property	Modifier	Type	Default value	Description	Inherited from	Defined in
`_tag`	`readonly`	`"@amqp-contract/TechnicalError"`	`undefined`	-	`TaggedError("@amqp-contract/TechnicalError", { name: "TechnicalError", })._tag`	node_modules/.pnpm/unthrown@5.12.2/node_modules/unthrown/dist/index.d.mts:2363
`cause?`	`public`	`unknown`	`undefined`	-	`MessageValidationError`.`cause`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es2022.error.d.ts:24
`message`	`public`	`string`	`undefined`	-	`TaggedError("@amqp-contract/TechnicalError", { name: "TechnicalError", }).message`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es5.d.ts:1075
`name`	`public`	`string`	`undefined`	-	`TaggedError("@amqp-contract/TechnicalError", { name: "TechnicalError", }).name`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es5.d.ts:1074
`stack?`	`public`	`string`	`undefined`	-	`TaggedError("@amqp-contract/TechnicalError", { name: "TechnicalError", }).stack`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es5.d.ts:1076
`tag`	`static readonly`	`"@amqp-contract/TechnicalError"`	`"@amqp-contract/TechnicalError"`	The `_tag`, for `P.tag(TechnicalError.tag)` without a raw string.	-	packages/core/src/errors.ts:45

Type Aliases

AmqpClientOptions

type AmqpClientOptions = ConnectionSource & object;

Defined in: packages/core/src/amqp-client.ts:186

Options for creating an AMQP client.

Type Declaration

Name	Type	Description	Defined in
`channelOptions?`	`Partial`<`CreateChannelOpts`>	-	packages/core/src/amqp-client.ts:197
`connectionPool?`	`string`	Partition of the connection pool this client draws from (ignored with an explicit `connection`). Clients with the same URLs, options AND pool share one TCP connection. `TypedAmqpClient` uses `"client"` and `TypedAmqpWorker` uses `"worker"`, so a publisher and a consumer in the same process never share a connection by default — RabbitMQ blocks a publishing connection under memory/disk alarms, and a consumer sharing it would stop acking with it. Defaults to `"default"`.	packages/core/src/amqp-client.ts:196
`connectTimeoutMs?`	`number` \| `null`	-	packages/core/src/amqp-client.ts:198
`logger?`	`Logger`	-	packages/core/src/amqp-client.ts:218
`publishTimeoutMs?`	`number` \| `null`	Maximum time in ms a publish may sit buffered waiting for the broker before its promise settles with a failure. Defaults to DEFAULT_PUBLISH_TIMEOUT_MS. Pass `null` to disable, restoring unbounded buffering — a publish issued during an outage then never settles. Same convention as `connectTimeoutMs`. Precedence against `channelOptions.publishTimeout` (most specific wins): 1. This option, when explicitly set to a number or `null` — always authoritative. 2. Otherwise `channelOptions.publishTimeout`, if the caller set one on the passthrough bag — respected, not silently overwritten. 3. Otherwise DEFAULT_PUBLISH_TIMEOUT_MS. `null` here disables the timeout outright, even when `channelOptions.publishTimeout` is also set — it is the more specific expression of intent.	packages/core/src/amqp-client.ts:217
`topology?`	`TopologyMode`	What the channel's setup does with the contract's topology on every (re)connect — see TopologyMode. Defaults to `"assert"`. The contract passed to the constructor is the scope: hand it a slice (e.g. `publisherTopology(contract)`) to declare only what one role needs.	packages/core/src/amqp-client.ts:225

AmqpConsumeOptions

type AmqpConsumeOptions = Omit<Options.Consume, "prefetch"> & object;

Defined in: packages/core/src/amqp-client.ts:261

Consume options that extend amqplib's Options.Consume with an optional per-consumer prefetch count.

Named AmqpConsumeOptions (not ConsumerOptions) so it never collides with the user-facing ConsumerOptions of @amqp-contract/worker.

prefetch maps to amqp-connection-manager's native per-consumer prefetch: it is applied via basic.qos(count, global=false) immediately before this consumer's basic.consume — and re-applied the same way when the consumer is re-established after a reconnect — so the value never bleeds onto other consumers sharing the channel.

Type Declaration

Name	Type	Description	Defined in
`prefetch?`	`number` \| `"unbounded"`	Per-consumer prefetch count, applied before `channel.consume(...)`. Defaults to DEFAULT_PREFETCH. Pass `"unbounded"` to opt out and let the broker push the entire ready backlog — AMQP's original default, and a memory hazard on any queue that can build a backlog. `"unbounded"` rather than `0` because AMQP's `0` means unlimited, which reads at a call site as its opposite.	packages/core/src/amqp-client.ts:272

AmqpPublishOptions

type AmqpPublishOptions = Options.Publish;

Defined in: packages/core/src/amqp-client.ts:246

Publish options for AmqpClient.publish / AmqpClient.sendToQueue.

Named AmqpPublishOptions (not PublishOptions) so it never collides with the user-facing PublishOptions of @amqp-contract/client.

Currently a re-export of amqplib's Options.Publish. A previous version of this type also exposed a timeout field, but that field never had a meaningful AMQP-level effect in this codebase and has been removed to avoid suggesting behaviour we do not provide. (amqp-connection-manager's own publishTimeout channel option is unrelated and is configured at channel creation, not per-publish.)

ConnectionSource

type ConnectionSource = object;

Defined in: packages/core/src/amqp-client.ts:145

Where a client's connection comes from — exactly one of (checked at construction; passing both or neither is a defect):

urls (+ optional connectionOptions): a connection dialled and pooled by amqp-contract, released when the last client using it closes. Multiple URLs provide failover.
connection: an AmqpConnectionManager the caller created and owns. The client only opens a channel on it; close() never closes it. Pass the same one to a client and a worker to share a TCP connection explicitly.

A flat type rather than a union so Omit<…>-style wrappers keep working.

Properties

Property	Type	Defined in
`connection?`	`AmqpConnectionManager`	packages/core/src/amqp-client.ts:148
`connectionOptions?`	`AmqpConnectionManagerOptions`	packages/core/src/amqp-client.ts:147
`urls?`	`ConnectionUrl`[]	packages/core/src/amqp-client.ts:146

ConsumeCallback

type ConsumeCallback = (msg) => void | Promise<void>;

Defined in: packages/core/src/amqp-client.ts:231

Callback type for consuming messages.

Parameters

Parameter	Type
`msg`	`ConsumeMessage` \| `null`

Returns

void | Promise<void>

Logger

type Logger = object;

Defined in: packages/core/src/logger.ts:30

Logger interface for amqp-contract packages.

Provides a simple logging abstraction that can be implemented by users to integrate with their preferred logging framework.

Example

typescript

// Simple console logger implementation
const logger: Logger = {
  debug: (message, context) => console.debug(message, context),
  info: (message, context) => console.info(message, context),
  warn: (message, context) => console.warn(message, context),
  error: (message, context) => console.error(message, context),
};

Methods

debug()

debug(message, context?): void;

Defined in: packages/core/src/logger.ts:36

Log debug level messages

Parameters

Parameter	Type	Description
`message`	`string`	The log message
`context?`	`LoggerContext`	Optional context to include with the log

Returns

void

error()

error(message, context?): void;

Defined in: packages/core/src/logger.ts:57

Log error level messages

Parameters

Parameter	Type	Description
`message`	`string`	The log message
`context?`	`LoggerContext`	Optional context to include with the log

Returns

void

info()

info(message, context?): void;

Defined in: packages/core/src/logger.ts:43

Log info level messages

Parameters

Parameter	Type	Description
`message`	`string`	The log message
`context?`	`LoggerContext`	Optional context to include with the log

Returns

void

warn()

warn(message, context?): void;

Defined in: packages/core/src/logger.ts:50

Log warning level messages

Parameters

Parameter	Type	Description
`message`	`string`	The log message
`context?`	`LoggerContext`	Optional context to include with the log

Returns

void

LoggerContext

type LoggerContext = Record<string, unknown> & object;

Defined in: packages/core/src/logger.ts:9

Context object for logger methods.

This type includes reserved keys that provide consistent naming for common logging context properties.

Type Declaration

Name	Type	Defined in
`error?`	`unknown`	packages/core/src/logger.ts:10

PublishFailureReason

type PublishFailureReason = "timeout" | "nacked" | "channel-closed";

Defined in: packages/core/src/errors.ts:98

Why the broker side of a publish failed — each one something core can actually observe on amqp-connection-manager's confirm channel:

"timeout" — no confirm arrived within publishTimeoutMs. Ambiguous: the message may have been written and be confirmed late, so republishing it can produce a duplicate. The worker's retry path accepts that (at-least-once): it requeues the original, and if the late copy did land in the wait queue, the handler runs twice.
"nacked" — the broker refused the message (basic.nack).
"channel-closed" — the channel closed before the message was confirmed.

TelemetryProvider

type TelemetryProvider = object;

Defined in: packages/core/src/telemetry.ts:55

Telemetry provider for AMQP operations. Uses lazy loading to gracefully handle cases where OpenTelemetry is not installed.

Properties

Property	Type	Description	Defined in
`getConsumeCounter`	() => `Counter` \| `undefined`	Get a counter for messages consumed. Returns undefined if OpenTelemetry is not available.	packages/core/src/telemetry.ts:72
`getConsumeLatencyHistogram`	() => `Histogram` \| `undefined`	Get a histogram for consume/process latency. Returns undefined if OpenTelemetry is not available.	packages/core/src/telemetry.ts:84
`getLateRpcReplyCounter`	() => `Counter` \| `undefined`	Get a counter for RPC replies that arrive after the caller has gone away (timeout, cancellation, or unknown correlationId). Returns undefined if OpenTelemetry is not available.	packages/core/src/telemetry.ts:91
`getPublishCounter`	() => `Counter` \| `undefined`	Get a counter for messages published. Returns undefined if OpenTelemetry is not available.	packages/core/src/telemetry.ts:66
`getPublishLatencyHistogram`	() => `Histogram` \| `undefined`	Get a histogram for publish latency. Returns undefined if OpenTelemetry is not available.	packages/core/src/telemetry.ts:78
`getRpcCallLatencyHistogram?`	() => `Histogram` \| `undefined`	Get a histogram for RPC round-trip duration (request publish → validated reply), kept apart from the publish histogram so a slow handler does not read as a slow broker. Optional so an existing custom provider keeps compiling; omitted means "not recorded".	packages/core/src/telemetry.ts:99
`getTracer`	() => `Tracer` \| `undefined`	Get a tracer instance for creating spans. Returns undefined if OpenTelemetry is not available.	packages/core/src/telemetry.ts:60

TopologyMode

type TopologyMode = "assert" | "passive" | "none";

Defined in: packages/core/src/setup.ts:19

What topology setup does with the contract's resources on (re)connect:

"assert" (default) — declare them (assertExchange / assertQueue, plus bindings): the broker creates what is missing and refuses a mismatch.
"passive" — only check they exist (checkExchange / checkQueue) and declare nothing, for an application whose credentials may not configure the broker; bindings cannot be checked passively and are skipped. A missing resource fails create().
"none" — touch nothing: topology is owned elsewhere (IaC, definitions import, another service).

Variables

DEFAULT_CONNECT_TIMEOUT_MS

const DEFAULT_CONNECT_TIMEOUT_MS: 30000 = 30_000;

Defined in: packages/core/src/amqp-client.ts:88

Default time waitForConnect will wait for the broker before erroring out. Defaulting to a finite value (rather than waiting forever) means a fail-fast developer experience: a misconfigured URL, a down broker, or wrong credentials surface as an err within 30 seconds. Pass null explicitly to disable the timeout.

DEFAULT_MAX_MESSAGE_BYTES

const DEFAULT_MAX_MESSAGE_BYTES: number;

Defined in: packages/core/src/codec.ts:27

Default cap on the size of a single inbound message body, after decompression. Applied to plain bodies as well as compressed ones: a few-KB compressed payload can otherwise expand to gigabytes before schema validation ever runs (a zip bomb), and an oversized plain body costs the same memory to parse. Matches RabbitMQ 4's default max_message_size. Over-cap messages surface through the defect channel and follow the poison-message DLQ path.

DEFAULT_PREFETCH

const DEFAULT_PREFETCH: 10 = 10;

Defined in: packages/core/src/amqp-client.ts:97

Default per-consumer prefetch.

Bounds in-flight messages per consumer, which bounds both memory and the redelivery burst when a worker crashes. Throughput-bound consumers raise it explicitly; "unbounded" restores AMQP's unlimited behavior.

DEFAULT_PUBLISH_TIMEOUT_MS

const DEFAULT_PUBLISH_TIMEOUT_MS: 30000 = 30_000;

Defined in: packages/core/src/amqp-client.ts:110

Default publishTimeout for the channel, in milliseconds.

Without a bound, publishes issued while the broker is unreachable buffer indefinitely and their promises never settle — a caller awaiting one waits forever. 30s is long enough that a brief reconnect does not fail healthy publishes, short enough that a real outage surfaces as an error.

Pass publishTimeoutMs: null to disable, matching the connectTimeoutMs convention.

defaultTelemetryProvider

const defaultTelemetryProvider: TelemetryProvider;

Defined in: packages/core/src/telemetry.ts:247

Default telemetry provider that uses OpenTelemetry API if available.

MessagingSemanticConventions

const MessagingSemanticConventions: object;

Defined in: packages/core/src/telemetry.ts:27

Semantic conventions for AMQP messaging following OpenTelemetry standards.

Type Declaration

Name	Type	Default value	Defined in
`AMQP_CONSUMER_NAME`	`"amqp.consumer.name"`	`"amqp.consumer.name"`	packages/core/src/telemetry.ts:38
`AMQP_PUBLISHER_NAME`	`"amqp.publisher.name"`	`"amqp.publisher.name"`	packages/core/src/telemetry.ts:37
`ERROR_TYPE`	`"error.type"`	`"error.type"`	packages/core/src/telemetry.ts:41
`MESSAGING_DESTINATION`	`"messaging.destination.name"`	`"messaging.destination.name"`	packages/core/src/telemetry.ts:30
`MESSAGING_DESTINATION_KIND`	`"messaging.destination.kind"`	`"messaging.destination.kind"`	packages/core/src/telemetry.ts:31
`MESSAGING_DESTINATION_KIND_EXCHANGE`	`"exchange"`	`"exchange"`	packages/core/src/telemetry.ts:45
`MESSAGING_DESTINATION_KIND_QUEUE`	`"queue"`	`"queue"`	packages/core/src/telemetry.ts:46
`MESSAGING_OPERATION`	`"messaging.operation"`	`"messaging.operation"`	packages/core/src/telemetry.ts:32
`MESSAGING_OPERATION_PROCESS`	`"process"`	`"process"`	packages/core/src/telemetry.ts:48
`MESSAGING_OPERATION_PUBLISH`	`"publish"`	`"publish"`	packages/core/src/telemetry.ts:47
`MESSAGING_RABBITMQ_MESSAGE_DELIVERY_TAG`	`"messaging.rabbitmq.message.delivery_tag"`	`"messaging.rabbitmq.message.delivery_tag"`	packages/core/src/telemetry.ts:36
`MESSAGING_RABBITMQ_ROUTING_KEY`	`"messaging.rabbitmq.destination.routing_key"`	`"messaging.rabbitmq.destination.routing_key"`	packages/core/src/telemetry.ts:35
`MESSAGING_SYSTEM`	`"messaging.system"`	`"messaging.system"`	packages/core/src/telemetry.ts:29
`MESSAGING_SYSTEM_RABBITMQ`	`"rabbitmq"`	`"rabbitmq"`	packages/core/src/telemetry.ts:44

See

https://opentelemetry.io/docs/specs/semconv/messaging/messaging-spans/

RPC_ERROR_CODE_HEADER

const RPC_ERROR_CODE_HEADER: "x-amqp-contract-error-code" = "x-amqp-contract-error-code";

Defined in: packages/core/src/errors.ts:213

AMQP message header carrying the error code of a typed RPC error reply.

A reply message with this header is an error reply: its body is { message, data } where data conforms to the error's declared schema in the RPC's errors map. A reply without it is a regular success reply whose body is the response payload — so success replies are wire-compatible with contracts that declare no errors.

Functions

isConnectionError()

function isConnectionError(error): error is ConnectionError;

Defined in: packages/core/src/errors.ts:193

Type guard to check if an error is a ConnectionError — the modeled failure of dialing the broker.

Parameters

Parameter	Type
`error`	`unknown`

Returns

error is ConnectionError

isMessageValidationError()

function isMessageValidationError(error): error is MessageValidationError;

Defined in: packages/core/src/errors.ts:200

Type guard to check if an error is a MessageValidationError.

Parameters

Parameter	Type
`error`	`unknown`

Returns

error is MessageValidationError

isRpcError()

function isRpcError(error): error is RpcError<string, unknown>;

Defined in: packages/core/src/errors.ts:260

Type guard to check if an error is an RpcError.

Narrowing to a specific code (and thus a typed data) is done on the code property after the guard, or via the error matcher on the _tag (matcher.with(P.tag("@amqp-contract/RpcError"), …)).

Parameters

Parameter	Type
`error`	`unknown`

Returns

error is RpcError<string, unknown>

isTechnicalError()

function isTechnicalError(error): error is TechnicalError;

Defined in: packages/core/src/errors.ts:185

Type guard to check if an error is a TechnicalError — the cause carried by every infrastructure Defect this library produces.

Parameters

Parameter	Type
`error`	`unknown`

Returns

error is TechnicalError

rpcError()

function rpcError<TCode, TData>(
   code, 
   data, 
   message?
): RpcError<TCode, TData>;

Defined in: packages/core/src/errors.ts:289

Create an RpcError with less verbosity.

The code/data pair must match one of the entries declared in the RPC's errors map — the handler's return type enforces this at compile time, and the worker validates data against the declared schema at runtime before replying.

Type Parameters

Type Parameter
`TCode` extends `string`
`TData`

Parameters

Parameter	Type	Description
`code`	`TCode`	The error code, as declared in the RPC's `errors` map
`data`	`TData`	The error data, validated against the declared schema
`message?`	`string`	Optional human-readable message (defaults to a generic one)

Returns

RpcError<TCode, TData>

Example

typescript

import { rpcError } from '@amqp-contract/worker';
import { ErrAsync } from 'unthrown';

const handler = ({ input: { payload } }) => {
  if (!orders.has(payload.orderId)) {
    return ErrAsync(rpcError('ORDER_NOT_FOUND', { orderId: payload.orderId }));
  }
  // ...
};

@amqp-contract/core ​

Classes ​

AmqpClient ​

Example ​

Constructors ​

Constructor ​

Parameters ​

Returns ​

Accessors ​

currentChannelEpoch ​

Get Signature ​

Returns ​

Methods ​

ack() ​

Parameters ​

Returns ​

addSetup() ​

Parameters ​

Returns ​

cancel() ​

Parameters ​

Returns ​

close() ​

Returns ​

consume() ​

Parameters ​

Returns ​

getConnection() ​

Returns ​

isConnected() ​

Returns ​

nack() ​

Parameters ​

Returns ​

on() ​

Parameters ​

Returns ​

publish() ​

Parameters ​

Returns ​

sendToQueue() ​

Parameters ​

Returns ​

waitForConnect() ​

Returns ​

ConnectionError ​

Extends ​

Constructors ​

Constructor ​

Parameters ​

Returns ​

Overrides ​

Properties ​

MessageValidationError ​

Param ​

Param ​

Extends ​

Constructors ​

Constructor ​

Parameters ​

Returns ​

Overrides ​

Properties ​

PublishError ​

Extends ​

Constructors ​

Constructor ​

Parameters ​

Returns ​

Overrides ​

Properties ​

RpcError ​

Extends ​

Type Parameters ​

Constructors ​

Constructor ​

Parameters ​

Returns ​

Overrides ​

Properties ​

@amqp-contract/core

Classes

AmqpClient

Example

Constructors

Constructor

Parameters

Returns

Accessors

currentChannelEpoch

Get Signature

Returns

Methods

ack()

Parameters

Returns

addSetup()

Parameters

Returns

cancel()

Parameters

Returns

close()

Returns

consume()

Parameters

Returns

getConnection()

Returns

isConnected()

Returns

nack()

Parameters

Returns

on()

Parameters

Returns

publish()

Parameters

Returns

sendToQueue()

Parameters

Returns

waitForConnect()

Returns

ConnectionError

Extends

Constructors

Constructor

Parameters

Returns

Overrides

Properties

MessageValidationError

Param

Param

Extends

Constructors

Constructor

Parameters

Returns

Overrides

Properties

PublishError

Extends

Constructors

Constructor

Parameters

Returns

Overrides

Properties

RpcError

Extends

Type Parameters

Constructors

Constructor

Parameters

Returns

Overrides

Properties