@btravstack/amqp-worker

---

@btravstack/amqp-worker

Classes

AmqpConfig

Defined in: amqp-worker/src/amqp-runtime.ts:49

The broker, as a service: amqp() 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: di/dist/index.d.mts:16

Returns

AmqpConfig

Inherited from

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

Properties

Property	Modifier	Type	Description	Inherited from	Defined in
[ID]	readonly	"AmqpConfig"	-	Port("AmqpConfig").[ID]	di/dist/index.d.mts:12
[SERVICE]	readonly	object	-	Port("AmqpConfig").[SERVICE]	di/dist/index.d.mts:13
[SERVICE].connectTimeoutMs	readonly	number	How long create waits for the connection before failing.	-	amqp-worker/src/amqp-runtime.ts:52
[SERVICE].url	readonly	string	-	-	amqp-worker/src/amqp-runtime.ts:50
portId	readonly	"AmqpConfig"	-	Port("AmqpConfig").portId	di/dist/index.d.mts:17

---

AmqpRuntime

Defined in: amqp-worker/src/amqp-runtime.ts:63

The runtime's port: what amqp() provides, and what the module start boots must export.

Extends

• RuntimePort<Runtime<never, AmqpInfo>>

Constructors

Constructor

new AmqpRuntime(): AmqpRuntime;

Defined in: di/dist/index.d.mts:16

Returns

AmqpRuntime

Inherited from

RuntimePort<Runtime<never, AmqpInfo>>.constructor

Properties

Property	Modifier	Type	Inherited from	Defined in
[ID]	readonly	"Runtime"	RuntimePort.[ID]	di/dist/index.d.mts:12
[SERVICE]	readonly	Runtime	RuntimePort.[SERVICE]	di/dist/index.d.mts:13
portId	readonly	"Runtime"	RuntimePort.portId	di/dist/index.d.mts:17

Type Aliases

AmqpConnectionOptions

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

Defined in: amqp-worker/src/amqp-runtime.ts:96

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: amqp-worker/src/amqp-runtime.ts:41

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

Properties

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

---

AmqpMessageOf

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

Defined in: amqp-worker/src/handler.ts:24

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: amqp-worker/src/handler.ts:29

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: amqp-worker/src/amqp-module.ts:49

Type Declaration

Name	Type	Description	Defined in
contract	TContract	-	amqp-worker/src/amqp-module.ts:63
exports?	X	The application's own exports; AmqpRuntime is added, since start resolves it.	amqp-worker/src/amqp-module.ts:76
handlers	Provider<HandlersInstanceOf<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.	amqp-worker/src/amqp-module.ts:70
imports?	I	-	amqp-worker/src/amqp-module.ts:73
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.	amqp-worker/src/amqp-module.ts:81
provides?	P	-	amqp-worker/src/amqp-module.ts:74

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[]

---

AmqpOptions

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

Defined in: amqp-worker/src/amqp-runtime.ts:122

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.	amqp-worker/src/amqp-runtime.ts:131

Type Parameters

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

---

HandlerPortOf

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

Defined in: amqp-worker/src/handler.ts:72

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: amqp-worker/src/amqp-runtime.ts:117

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: amqp-worker/src/amqp-runtime.ts:111

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

Type Parameters

Type Parameter
C extends AnyAmqpContract

Functions

amqp()

function amqp<TContract, Unit>(options): Module<AmqpRuntime | AmqpConfig, ConfigInvalid, 
  | Env
  | HandlersInstanceOf<TContract>
  | UnitNeedsOf<Unit>>;

Defined in: amqp-worker/src/amqp-runtime.ts:171

The AMQP starter: a module providing the runtime and its configuration (bound from AMQP_URL unless pinned), built over the handlers the application provides on the starter's own handlers port — the module's one need. Import it next to the application, provide the handlers, export AmqpRuntime.

With url pinned the module reads nothing from the environment.

Type Parameters

Type Parameter	Default type
TContract extends ContractDefinition	-
Unit extends AnyUnitModule | undefined	undefined

Parameters

Parameter	Type
options	AmqpOptions<TContract, Unit>

Returns

Module<AmqpRuntime | AmqpConfig, ConfigInvalid, | Env | HandlersInstanceOf<TContract> | UnitNeedsOf<Unit>>

---

AmqpHandler()

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

Defined in: amqp-worker/src/handler.ts:108

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: amqp-worker/src/amqp-runtime.ts:337

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: amqp-worker/src/handler.ts:62

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, 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: amqp-worker/src/amqp-module.ts:110

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