@amqp-contract/core

---

@amqp-contract/core

Classes

AmqpClient

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

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

All operations return AsyncResult<T, never>: infrastructure failures are unexpected, so they surface through the Defect channel (with a TechnicalError as the defect's cause for logging), never as a modeled Err.

Example

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:275

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:568

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:607

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:649

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

This is useful for setting up channel-level configuration like prefetch.

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:556

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:682

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:518

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:371

Get the underlying connection manager

This method exposes the AmqpConnectionManager instance that this client uses. The connection is automatically shared across all AmqpClient instances that use the same URLs and connection options.

Returns

IAmqpConnectionManager

The AmqpConnectionManager instance used by this client

nack()

nack(msg, options?): void;

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

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:664

Register an event listener on the channel wrapper.

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, never>;

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

Publish a message to an exchange.

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

A full channel write buffer (the wrapper's boolean false confirmation) surfaces as a Defect with a TechnicalError cause — like every other publish-side infrastructure failure. Callers never see the boolean.

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, never>

sendToQueue()

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

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

Publish a message directly to a queue.

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

A full channel write buffer surfaces as a Defect with a TechnicalError cause — see publish.

Parameters

Parameter	Type
queue	string
content	unknown
options?	Publish

Returns

AsyncResult<void, never>

waitForConnect()

waitForConnect(): AsyncResult<void, never>;

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

Wait for the channel to be connected and ready.

If connectTimeoutMs was provided in the constructor options, the returned AsyncResult resolves to a Defect (a TechnicalError cause) once the timeout elapses. Without a timeout, 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, never>

---

MessageValidationError

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

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:52

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	Inherited from	Defined in
_tag	readonly	"@amqp-contract/MessageValidationError"	TaggedError("@amqp-contract/MessageValidationError", { name: "MessageValidationError", })._tag	node_modules/.pnpm/unthrown@5.1.0/node_modules/unthrown/dist/index.d.mts:1941
cause?	public	unknown	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	TaggedError("@amqp-contract/MessageValidationError", { name: "MessageValidationError", }).issues	packages/core/src/errors.ts:50
message	public	string	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	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	TaggedError("@amqp-contract/MessageValidationError", { name: "MessageValidationError", }).source	packages/core/src/errors.ts:49
stack?	public	string	TaggedError("@amqp-contract/MessageValidationError", { name: "MessageValidationError", }).stack	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es5.d.ts:1076

---

RpcError

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

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:126

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	Overrides	Inherited from	Defined in
_tag	readonly	"@amqp-contract/RpcError"	-	TaggedError( "@amqp-contract/RpcError", { name: "RpcError" }, )._tag	node_modules/.pnpm/unthrown@5.1.0/node_modules/unthrown/dist/index.d.mts:1941
cause?	public	unknown	-	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	TaggedError( "@amqp-contract/RpcError", { name: "RpcError" }, ).code	-	packages/core/src/errors.ts:123
data	readonly	TData	TaggedError( "@amqp-contract/RpcError", { name: "RpcError" }, ).data	-	packages/core/src/errors.ts:124
message	public	string	-	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	-	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	-	TaggedError( "@amqp-contract/RpcError", { name: "RpcError" }, ).stack	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es5.d.ts:1076

---

TechnicalError

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

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

This includes AMQP connection failures, channel issues, compression/parse faults, and other unexpected runtime errors. Shared across core, worker, and client packages.

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:29

Parameters

Parameter	Type
message	string
cause?	unknown

Returns

TechnicalError

Overrides

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

Properties

Property	Modifier	Type	Inherited from	Defined in
_tag	readonly	"@amqp-contract/TechnicalError"	TaggedError("@amqp-contract/TechnicalError", { name: "TechnicalError", })._tag	node_modules/.pnpm/unthrown@5.1.0/node_modules/unthrown/dist/index.d.mts:1941
cause?	public	unknown	MessageValidationError.cause	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es2022.error.d.ts:24
message	public	string	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	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	TaggedError("@amqp-contract/TechnicalError", { name: "TechnicalError", }).stack	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es5.d.ts:1076

Type Aliases

AmqpClientOptions

type AmqpClientOptions = object;

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

Options for creating an AMQP client.

Properties

Property	Type	Description	Defined in
channelOptions?	Partial<CreateChannelOpts>	Optional channel configuration options.	packages/core/src/amqp-client.ts:136
connectionOptions?	AmqpConnectionManagerOptions	Optional connection configuration (heartbeat, reconnect settings, etc.).	packages/core/src/amqp-client.ts:135
connectTimeoutMs?	number | null	Maximum time in ms to wait for the channel to become ready in waitForConnect. Defaults to DEFAULT_CONNECT_TIMEOUT_MS. Pass null to disable the timeout entirely (amqp-connection-manager will retry indefinitely).	packages/core/src/amqp-client.ts:137
logger?	Logger	Optional logger. Channel-level 'error' events (topology setup failures on connect/reconnect, publish-worker faults) are routed here — they are recoverable-by-reconnect conditions, never thrown.	packages/core/src/amqp-client.ts:157
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. See the field's own doc comment for the precedence against channelOptions.publishTimeout.	packages/core/src/amqp-client.ts:156
urls	ConnectionUrl[]	AMQP broker URL(s). Multiple URLs provide failover support.	packages/core/src/amqp-client.ts:134

---

AmqpConsumeOptions

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

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

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:204

---

AmqpPublishOptions

type AmqpPublishOptions = Options.Publish;

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

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.)

---

ConnectionLease

type ConnectionLease = object;

Defined in: packages/core/src/connection-manager.ts:10

A held reference to a pooled connection.

Properties

Property	Type	Description	Defined in
connection	AmqpConnectionManager	The shared connection this lease holds a reference to.	packages/core/src/connection-manager.ts:12
release	() => Promise<void>	Release this lease. Idempotent — a double release() (e.g. a client whose close() runs on both a finally and an error path) is a no-op, so it can never underflow the pool's reference count and close a connection out from under another live client.	packages/core/src/connection-manager.ts:19

---

ConsumeCallback

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

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

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

// 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

---

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
getTracer	() => Tracer | undefined	Get a tracer instance for creating spans. Returns undefined if OpenTelemetry is not available.	packages/core/src/telemetry.ts:60

Variables

DEFAULT_CONNECT_TIMEOUT_MS

const DEFAULT_CONNECT_TIMEOUT_MS: 30000 = 30_000;

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

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_PREFETCH

const DEFAULT_PREFETCH: 10 = 10;

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

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:92

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:230

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:96

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

endSpanError()

function endSpanError(span, error): void;

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

End a span with error status. Never throws.

Parameters

Parameter	Type
span	Span | undefined
error	Error

Returns

void

---

endSpanSuccess()

function endSpanSuccess(span): void;

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

End a span with success status. Never throws.

Parameters

Parameter	Type
span	Span | undefined

Returns

void

---

isMessageValidationError()

function isMessageValidationError(error): error is MessageValidationError;

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

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:139

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:76

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

---

recordConsumeMetric()

function recordConsumeMetric(
   provider, 
   queueName, 
   consumerName, 
   success, 
   durationMs): void;

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

Record a consume metric. Never throws.

Parameters

Parameter	Type
provider	TelemetryProvider
queueName	string
consumerName	string
success	boolean
durationMs	number

Returns

void

---

recordLateRpcReply()

function recordLateRpcReply(provider, reason): void;

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

Record an RPC reply that arrived after the caller stopped waiting.

Parameters

Parameter	Type	Description
provider	TelemetryProvider	-
reason	"unknown-correlation-id" | "missing-correlation-id"	Why the reply was orphaned. "unknown-correlation-id" is the typical "caller already timed out" case; "missing-correlation-id" means the broker delivered a reply with no correlationId at all (a protocol violation by the responder).

Returns

void

---

recordPublishMetric()

function recordPublishMetric(
   provider, 
   exchangeName, 
   routingKey, 
   success, 
   durationMs): void;

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

Record a publish metric. Never throws.

Parameters

Parameter	Type
provider	TelemetryProvider
exchangeName	string
routingKey	string | undefined
success	boolean
durationMs	number

Returns

void

---

rpcError()

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

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

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

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

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

---

safeJsonParse()

function safeJsonParse<R>(buffer, qualify): Result<unknown, Exclude<R, Defect>>;

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

Parse a Buffer as JSON, triaging any JSON.parse exception through the caller-supplied qualify callback.

Use this in consume / reply paths where a parse failure is a typed value, not a thrown exception — the caller decides how to translate the raw error into a domain-level error (e.g. TechnicalError), or routes it to the defect channel via the injected defect helper (the full unthrown qualify signature, so no model-then-defect round-trip is ever needed).

Type Parameters

Type Parameter	Description
R	What the qualify callback produces: a modeled error type, the defect marker, or a union of both. The defect marker is subtracted from the resulting error channel, exactly like fromThrowable.

Parameters

Parameter	Type	Description
buffer	Buffer	The raw message body to parse.
qualify	(raw, defect) => R & NotThenable<R>	Callback invoked with the underlying JSON.parse error and the injected defect helper; returns the modeled error or defect(cause).

Returns

Result<unknown, Exclude<R, Defect>>

A Result containing the parsed unknown value or the mapped error.

Examples

Modeled error

const parsed = safeJsonParse(
  msg.content,
  (error) => new TechnicalError("Failed to parse JSON", error),
);

Defect-channel routing

const parsed = safeJsonParse(
  msg.content,
  (error, defect) => defect(new TechnicalError("Failed to parse JSON", error)),
); // Result<unknown, never>

---

setupAmqpTopology()

function setupAmqpTopology(channel, contract): Promise<void>;

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

Setup AMQP topology (exchanges, queues, and bindings) from a contract definition.

This function sets up the complete AMQP topology in the correct order:

1. Assert all exchanges defined in the contract
2. Validate dead letter exchanges are declared before referencing them
3. Assert all queues with their configurations (including dead letter settings), plus the TTL-backoff wait queues derived from each queue's retry config (one per distinct backoff delay — see deriveTtlBackoffInfrastructure)
4. Create all bindings (queue-to-exchange and exchange-to-exchange)

Parameters

Parameter	Type	Description
channel	Channel	The AMQP channel to use for topology setup
contract	ContractDefinition	The contract definition containing the topology specification

Returns

Promise<void>

Throws

If any exchanges, queues, or bindings fail to be created

Throws

If a queue references a dead letter exchange not declared in the contract

Example

const channel = await connection.createChannel();
await setupAmqpTopology(channel, contract);

---

startConsumeSpan()

function startConsumeSpan(
   provider, 
   queueName, 
   consumerName, 
   attributes?): Span | undefined;

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

Create a span for a consume/process operation. Returns undefined if OpenTelemetry is not available. Never throws — a throwing provider is treated as "no telemetry".

Parameters

Parameter	Type
provider	TelemetryProvider
queueName	string
consumerName	string
attributes?	Attributes

Returns

Span | undefined

---

startPublishSpan()

function startPublishSpan(
   provider, 
   exchangeName, 
   routingKey, 
   attributes?): Span | undefined;

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

Create a span for a publish operation. Returns undefined if OpenTelemetry is not available. Never throws — a throwing provider is treated as "no telemetry".

Parameters

Parameter	Type
provider	TelemetryProvider
exchangeName	string
routingKey	string | undefined
attributes?	Attributes

Returns

Span | undefined

---

technicalDefect()

function technicalDefect(error): Result<never, never>;

Defined in: packages/core/src/defect.ts:20

Mint a Defect-carrying Result from a TechnicalError, for the imperative sites (outside a combinator callback) that must surface an unexpected infrastructure failure through the defect channel. Uses the fromSafeThrowable boundary — the sanctioned way to route a throw to a Defect without a public defect constructor.

The shape is the sync Result<never, never>: it fits every channel (both T and E are never), so it can seed any pipeline. Call .toAsync() where an AsyncResult is expected.

Shared by @amqp-contract/client and @amqp-contract/worker (each used to hand-roll its own copy); exported so any layer can mint a defect the same way instead of re-deriving the boundary trick.

Parameters

Parameter	Type
error	TechnicalError

Returns

Result<never, never>