@btravstack/amqp-worker

@btravstack/amqp-worker

Classes

AmqpConfig

Defined in: packages/amqp-worker/src/amqp-runtime.ts:54

The broker, as a service: the starter binds it from AMQP_URL (default amqp://127.0.0.1:5672) and AMQP_CONNECT_TIMEOUT_MS (default 5_000) unless pinned, and anything else in the graph may read it — a publisher sharing the consumer's broker, say.

Extends

PortInstance<"AmqpConfig", { connectTimeoutMs: number; url: string; }>

Constructors

Constructor

new AmqpConfig(): AmqpConfig;

Defined in: packages/di/dist/index.d.mts:15

Returns

AmqpConfig

Inherited from

Port("AmqpConfig")<{ readonly url: string; /** How long `create` waits for the connection and topology setup before failing. */ readonly connectTimeoutMs: number; }>.constructor

Properties

Property	Modifier	Type	Description	Inherited from	Defined in
`[ID]`	`readonly`	`"AmqpConfig"`	-	`Port("AmqpConfig").[ID]`	packages/di/dist/index.d.mts:11
`[SERVICE]`	`readonly`	`object`	-	`Port("AmqpConfig").[SERVICE]`	packages/di/dist/index.d.mts:12
`[SERVICE].connectTimeoutMs`	`readonly`	`number`	How long `create` waits for the connection and topology setup before failing.	-	packages/amqp-worker/src/amqp-runtime.ts:57
`[SERVICE].url`	`readonly`	`string`	-	-	packages/amqp-worker/src/amqp-runtime.ts:55
`portId`	`static readonly`	`"AmqpConfig"`	-	`Port("AmqpConfig").portId`	packages/di/dist/index.d.mts:16

AmqpRuntime

Defined in: packages/amqp-worker/src/amqp-runtime.ts:68

The runtime's port: what the starter provides, and what AmqpModule exports for start to boot.

Extends

RuntimePort<Runtime<never, AmqpInfo>>

Constructors

Constructor

new AmqpRuntime(): AmqpRuntime;

Defined in: packages/di/dist/index.d.mts:15

Returns

AmqpRuntime

Inherited from

RuntimePort<Runtime<never, AmqpInfo>>.constructor

Properties

Property	Modifier	Type	Inherited from	Defined in
`[ID]`	`readonly`	`"Runtime"`	`RuntimePort.[ID]`	packages/di/dist/index.d.mts:11
`[SERVICE]`	`readonly`	`Runtime`	`RuntimePort.[SERVICE]`	packages/di/dist/index.d.mts:12
`portId`	`static readonly`	`"Runtime"`	`RuntimePort.portId`	packages/di/dist/index.d.mts:16

InvalidAmqpSubscription

Defined in: packages/amqp-worker/src/subscription.ts:36

A subscription requires explicitly ephemeral queues with no retry or dead-letter routing.

Extends

TaggedErrorInstance<"InvalidAmqpSubscription", { subscription: string; }>

Constructors

Constructor

new InvalidAmqpSubscription(args): InvalidAmqpSubscription;

Defined in: node_modules/.pnpm/unthrown@5.12.1/node_modules/unthrown/dist/index.d.mts:2386

Parameters

Parameter	Type
`args`	`object` & `object`

Returns

InvalidAmqpSubscription

Inherited from

TaggedError("InvalidAmqpSubscription")<{ readonly subscription: string; }>.constructor

Properties

Property	Modifier	Type	Default value	Overrides	Inherited from	Defined in
`_tag`	`readonly`	`"InvalidAmqpSubscription"`	`undefined`	-	`TaggedError("InvalidAmqpSubscription")._tag`	node_modules/.pnpm/unthrown@5.12.1/node_modules/unthrown/dist/index.d.mts:2363
`cause?`	`public`	`unknown`	`undefined`	-	`TaggedError("InvalidAmqpSubscription").cause`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es2022.error.d.ts:24
`message`	`public`	`string`	`"subscriptions require consumers on exclusive auto-delete non-durable classic queues with no retry or dead letter, and no RPCs"`	`TaggedError("InvalidAmqpSubscription").message`	-	packages/amqp-worker/src/subscription.ts:39
`name`	`public`	`string`	`undefined`	-	`TaggedError("InvalidAmqpSubscription").name`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es5.d.ts:1074
`stack?`	`public`	`string`	`undefined`	-	`TaggedError("InvalidAmqpSubscription").stack`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es5.d.ts:1076
`subscription`	`readonly`	`string`	`undefined`	-	`TaggedError("InvalidAmqpSubscription").subscription`	packages/amqp-worker/src/subscription.ts:37

Type Aliases

AmqpConnectionOptions

type AmqpConnectionOptions = NonNullable<Parameters<typeof TypedAmqpWorker.create>[0]["connectionOptions"]>;

Defined in: packages/amqp-worker/src/amqp-runtime.ts:85

The connection tuning TypedAmqpWorker.create accepts — heartbeat, reconnect interval, findServers, TLS/socket options. The library declares it (AmqpConnectionManagerOptions) without exporting it by name, so it is reached by index for the same reason AnyAmqpContract is.

AmqpInfo

type AmqpInfo = object;

Defined in: packages/amqp-worker/src/amqp-runtime.ts:46

What the worker publishes once it is consuming, read back through RunningApp.runtimeInfo().

Properties

Property	Modifier	Type	Defined in
`queues`	`readonly`	readonly `string`[]	packages/amqp-worker/src/amqp-runtime.ts:46

AmqpMessageOf

type AmqpMessageOf<C> = Parameters<Extract<WorkerInferHandlers<C>[HandlerKeyOf<C>], (...args) => unknown>>[1];

Defined in: packages/amqp-worker/src/handler.ts:25

The validated delivery, typed by the contract: the union of every consumer's and rpc's own consumed message. Reached through the handler's own second parameter, since @amqp-contract/worker exports WorkerInferConsumedMessage but not the InferConsumerNames constraint it takes — the same by-index route AnyAmqpContract travels.

Type Parameters

Type Parameter
`C` extends `AnyAmqpContract`

AmqpMessagePortOf

type AmqpMessagePortOf<C> = PortClassOf<"AmqpMessage", AmqpMessageOf<C>>;

Defined in: packages/amqp-worker/src/handler.ts:30

The seeded port's class, typed for C: what AmqpMessage(contract) answers.

Type Parameters

Type Parameter
`C` extends `AnyAmqpContract`

AmqpModuleOptions

type AmqpModuleOptions<TContract, HandlersError, HandlersNeeds, Declared, Unit, I, P, X, N> = AmqpTuning<Unit> & object & NeedsGate<Imports<I, TContract, Unit>, Provides<P, TContract, HandlersError, HandlersNeeds>, N> & UnitGate<Unit, Declared>;

Defined in: packages/amqp-worker/src/amqp-module.ts:51

Type Parameters

Type Parameter
`TContract` extends `AnyAmqpContract`
`HandlersError`
`HandlersNeeds`
`Declared`
`Unit` extends `AnyUnitModule` \| `undefined`
`I` extends readonly `AnyModule`[]
`P` extends readonly `AnyProvider`[]
`X` extends readonly `Exportable`<`Imports`<`I`, `TContract`, `Unit`>, `Provides`<`P`, `TContract`, `HandlersError`, `HandlersNeeds`>>[]
`N` extends readonly `AnyPort`[] \| `undefined`

Type Declaration

Name	Type	Description	Defined in
`contract`	`TContract`	-	packages/amqp-worker/src/amqp-module.ts:65
`exports?`	`X`	The application's own exports; `AmqpRuntime` and visible kernel infrastructure are added.	packages/amqp-worker/src/amqp-module.ts:82
`handlers`	`Provider`<`HandlersInstanceOf`<`NoInfer`<`TContract`>>, `HandlersError`, `HandlersNeeds`> & `object`	The application's handlers — what `AmqpHandlers(contract)(…)` returns for THIS contract. Its `_declaredUnit` phantom carries every port its pieces read off `context.unit`, which is what UnitGate checks `unit.message` against.	packages/amqp-worker/src/amqp-module.ts:72
`imports?`	`I`	-	packages/amqp-worker/src/amqp-module.ts:79
`needs?`	`N`	What this root's OWN providers expect from outside. di's gate is re-stated over the augmented tuples below, so forgetting one is an error at THIS call.	packages/amqp-worker/src/amqp-module.ts:87
`provides?`	`P`	-	packages/amqp-worker/src/amqp-module.ts:80

AmqpOptions

type AmqpOptions<TContract, Unit> = object & AmqpTuning<Unit>;

Defined in: packages/amqp-worker/src/amqp-runtime.ts:111

Type Parameters

Type Parameter	Default type
`TContract` extends `AnyAmqpContract`	-
`Unit` extends `AnyUnitModule` \| `undefined`	`undefined`

Type Declaration

Name	Type	Description	Defined in
`contract`	`TContract`	The contract; the handlers port is typed by it — one handler per `consumers` / `rpcs` key, with no injected context, since a handler is built by di from the services it declares.	packages/amqp-worker/src/amqp-runtime.ts:120

HandlerPortOf

type HandlerPortOf<C, K> = PortClassOf<`${typeof HANDLER_PREFIX}${K}`, WorkerInferHandlers<C>[K]>;

Defined in: packages/amqp-worker/src/handler.ts:73

The port one piece targets. Its id carries the contract key, which is what makes two slices claiming one consumer di's duplicate-provider defect rather than a silent merge.

Type Parameters

Type Parameter
`C` extends `AnyAmqpContract`
`K` extends `HandlerKeyOf`<`C`>

HandlersInstanceOf

type HandlersInstanceOf<C> = PortInstance<"AmqpHandlers", WorkerInferHandlers<C>>;

Defined in: packages/amqp-worker/src/amqp-runtime.ts:106

The handlers port's instance for C — the module's one need.

Type Parameters

Type Parameter
`C` extends `AnyAmqpContract`

HandlersPortOf

type HandlersPortOf<C> = PortClassOf<"AmqpHandlers", WorkerInferHandlers<C>>;

Defined in: packages/amqp-worker/src/amqp-runtime.ts:100

The handlers port class, typed for C: what AmqpHandlers(contract)(…).port is.

Type Parameters

Type Parameter
`C` extends `AnyAmqpContract`

SubscriptionHandlers

type SubscriptionHandlers<C, E> = { readonly [K in C["consumers"] extends Record<string, unknown> ? keyof C["consumers"] : never]: (args: Parameters<WorkerInferConsumerHandler<C, K>>) => void | Result<void, E> | AsyncResult<void, E> };

Defined in: packages/amqp-worker/src/subscription.ts:44

Best-effort handlers retain their contract's typed input and may run synchronously.

Type Parameters

Type Parameter
`C` extends `AnyAmqpContract`
`E`

Functions

AmqpHandler()

function AmqpHandler<C, K>(contract, key): <D, U>(options) => MintedHandler<C, K, InstanceType<D[keyof D]>, U>;

Defined in: packages/amqp-worker/src/handler.ts:107

One consumer or rpc of a contract, as a provider on a port of its own.

A worker with several consumers is several pieces, each declaring the services its own handler calls; AmqpHandlers(contract)([...]) composes them. contract is read for its TYPE only, and types both key and the handler — so a consumer the contract does not declare is a compile error here.

There is no name to give: the contract key IS the port's name.

unit declares the ports the handler reads off context.unit, resolved out of the fork the delivery opened; the root's unit.message module must export every one of them.

Type Parameters

Type Parameter
`C` extends `ContractDefinition`
`K` extends `string`

Parameters

Parameter	Type
`contract`	`C`
`key`	`K`

Returns

<D, U>(options) => MintedHandler<C, K, InstanceType<D[keyof D]>, U>

AmqpHandlers()

function AmqpHandlers<C>(contract): Whole<C> & Compose<C>;

Defined in: packages/amqp-worker/src/amqp-runtime.ts:312

The handlers as a provider, from the contract. Two call forms, one port.

AmqpHandlers(orderContract)({ inject: { logger: Logger }, sync: ({ logger }) => ({ orderNotifications: (m) => … }) })
AmqpHandlers(orderContract)([orderNotifications, orderAudit])

The first is { inject?, unit?, sync }, whose sync hands back the whole handlers record — one unit: record for every entry in it. The second takes the pieces AmqpHandler(contract, key) builds: they are the provider's deps, keyed by the contract key each piece's port id carries, so the services record IS the handlers record. Every declared key must be covered, and two slices claiming one key are di's duplicate-provider defect at build.

Type Parameters

Type Parameter
`C` extends `ContractDefinition`

Parameters

Parameter	Type
`contract`	`C`

Returns

Whole<C> & Compose<C>

AmqpMessage()

function AmqpMessage<C>(contract): AmqpMessagePortOf<C>;

Defined in: packages/amqp-worker/src/handler.ts:63

The validated delivery as a port: the one thing the worker seeds the fork with, so a unit.message module derives a tenant — or anything else — from the message rather than from an ambient record.

const Message = AmqpMessage(orderContract);
const MessageUnit = Module("MessageUnit")({
  needs: [Message],
  provides: [Provider(Tenant)({ inject: { message: Message }, sync: ({ message }) => message.payload.tenantId })],
  exports: [Tenant],
});

contract is read for its TYPE only. One Port(...) call fixed per contract at the type level, the move AmqpHandlersPort makes, so a module built for one contract cannot read another's message.

Type Parameters

Type Parameter
`C` extends `ContractDefinition`

Parameters

Parameter	Type
`contract`	`C`

Returns

AmqpMessagePortOf<C>

AmqpModule()

function AmqpModule<Name>(name): <TContract, HandlersError, HandlersNeeds, Declared, Unit, I, P, X, N>(options) => Module<ResolvedExports<readonly [typeof AmqpRuntime, 
  | Logger extends Available<readonly [I, AmqpStarter<TContract, Unit>], readonly [Provider<HandlersInstanceOf<TContract>, HandlersError, HandlersNeeds>, P]> ? typeof Logger : never
  | HealthChecks extends Available<readonly [I, AmqpStarter<TContract, Unit>], readonly [Provider<HandlersInstanceOf<TContract>, HandlersError, HandlersNeeds>, P]> ? typeof HealthChecks : never, X]>, 
  | ErrOf<readonly [Provider<HandlersInstanceOf<TContract>, HandlersError, HandlersNeeds>, P][number]>
  | ErrOfModule<readonly [I, AmqpStarter<TContract, Unit>][number]>, 
  | Exclude<NeedOf<readonly [Provider<HandlersInstanceOf<TContract>, HandlersError, HandlersNeeds>, P][number]>, Available<readonly [I, AmqpStarter<TContract, Unit>], readonly [Provider<HandlersInstanceOf<TContract>, HandlersError, HandlersNeeds>, P]>>
  | Exclude<NeedsOfModule<readonly [I, AmqpStarter<TContract, Unit>][number]>, Available<readonly [I, AmqpStarter<TContract, Unit>], readonly [Provider<HandlersInstanceOf<TContract>, HandlersError, HandlersNeeds>, P]>>>;

Defined in: packages/amqp-worker/src/amqp-module.ts:116

Module(name)({...}) for an AMQP deployment: everything a di module takes, plus the contract and the handlers provider. The sugar imports the starter, provides the handlers and exports AmqpRuntime, handing back exactly the module Module(...) would have declared over the augmented tuples.

export const OrderAmqpWorker = AmqpModule("OrderAmqpWorker")({
  contract: orderContract,
  handlers: orderHandlers,
  imports: [OrderApplicationModule, OrderPersistenceModule],
  exports: [Logger],
});
await runMain(OrderAmqpWorker);

Type Parameters

Type Parameter
`Name` extends `string`

Parameters

Parameter	Type
`name`	`Name`

Returns

<TContract, HandlersError, HandlersNeeds, Declared, Unit, I, P, X, N>(options) => Module<ResolvedExports<readonly [typeof AmqpRuntime, | Logger extends Available<readonly [I, AmqpStarter<TContract, Unit>], readonly [Provider<HandlersInstanceOf<TContract>, HandlersError, HandlersNeeds>, P]> ? typeof Logger : never | HealthChecks extends Available<readonly [I, AmqpStarter<TContract, Unit>], readonly [Provider<HandlersInstanceOf<TContract>, HandlersError, HandlersNeeds>, P]> ? typeof HealthChecks : never, X]>, | ErrOf<readonly [Provider<HandlersInstanceOf<TContract>, HandlersError, HandlersNeeds>, P][number]> | ErrOfModule<readonly [I, AmqpStarter<TContract, Unit>][number]>, | Exclude<NeedOf<readonly [Provider<HandlersInstanceOf<TContract>, HandlersError, HandlersNeeds>, P][number]>, Available<readonly [I, AmqpStarter<TContract, Unit>], readonly [Provider<HandlersInstanceOf<TContract>, HandlersError, HandlersNeeds>, P]>> | Exclude<NeedsOfModule<readonly [I, AmqpStarter<TContract, Unit>][number]>, Available<readonly [I, AmqpStarter<TContract, Unit>], readonly [Provider<HandlersInstanceOf<TContract>, HandlersError, HandlersNeeds>, P]>>>

amqpSubscription()

function amqpSubscription<C, Settings, SettingsId, E, D>(name, options): Module<HealthChecks, ConnectionError | InvalidAmqpSubscription, 
  | Scope
  | PortInstance<SettingsId, Settings>
  | InstanceType<D[keyof D]>>;

Defined in: packages/amqp-worker/src/subscription.ts:55

Consume broadcast facts for the application scope's lifetime, beside its one runtime. Every validated delivery is acknowledged, including handler failures; failures are reported to Observers. Each replica must name its own queues.

Type Parameters

Type Parameter	Default type
`C` extends `ContractDefinition`	-
`Settings` extends `object`	-
`SettingsId` extends `string`	-
`E`	-
`D` extends `Readonly`<`Record`<`string`, `AnyPort`>>	`Record`<`never`, `never`>

Parameters

Parameter	Type
`name`	`string`
`options`	`object` & ([keyof `D`] extends [`never`] ? `unknown` : `object`)

Returns

Module<HealthChecks, ConnectionError | InvalidAmqpSubscription, | Scope | PortInstance<SettingsId, Settings> | InstanceType<D[keyof D]>>

@btravstack/amqp-worker ​

Classes ​

AmqpConfig ​

Extends ​

Constructors ​

Constructor ​

Returns ​

Inherited from ​

Properties ​

AmqpRuntime ​

Extends ​

Constructors ​

Constructor ​

Returns ​

Inherited from ​

Properties ​

InvalidAmqpSubscription ​

Extends ​

Constructors ​

Constructor ​

Parameters ​

Returns ​

Inherited from ​

Properties ​

Type Aliases ​

AmqpConnectionOptions ​

AmqpInfo ​

Properties ​

AmqpMessageOf ​

Type Parameters ​

AmqpMessagePortOf ​

Type Parameters ​

AmqpModuleOptions ​

Type Parameters ​

Type Declaration ​

AmqpOptions ​

Type Parameters ​

Type Declaration ​

HandlerPortOf ​

Type Parameters ​

HandlersInstanceOf ​

Type Parameters ​

HandlersPortOf ​

Type Parameters ​

SubscriptionHandlers ​

Type Parameters ​

Functions ​

AmqpHandler() ​

Type Parameters ​

Parameters ​

Returns ​

AmqpHandlers() ​

Type Parameters ​

Parameters ​

Returns ​

AmqpMessage() ​

Type Parameters ​

Parameters ​

Returns ​

AmqpModule() ​

Type Parameters ​

Parameters ​

Returns ​

amqpSubscription() ​

Type Parameters ​

Parameters ​

Returns ​

@btravstack/amqp-worker

Classes

AmqpConfig

Extends

Constructors

Constructor

Returns

Inherited from

Properties

AmqpRuntime

Extends

Constructors

Constructor

Returns

Inherited from

Properties

InvalidAmqpSubscription

Extends

Constructors

Constructor

Parameters

Returns

Inherited from

Properties

Type Aliases

AmqpConnectionOptions

AmqpInfo

Properties

AmqpMessageOf

Type Parameters

AmqpMessagePortOf

Type Parameters

AmqpModuleOptions

Type Parameters

Type Declaration

AmqpOptions

Type Parameters

Type Declaration

HandlerPortOf

Type Parameters

HandlersInstanceOf

Type Parameters

HandlersPortOf

Type Parameters

SubscriptionHandlers

Type Parameters

Functions

AmqpHandler()

Type Parameters

Parameters

Returns

AmqpHandlers()

Type Parameters

Parameters

Returns

AmqpMessage()

Type Parameters

Parameters

Returns

AmqpModule()

Type Parameters

Parameters

Returns

amqpSubscription()

Type Parameters

Parameters

Returns