Contract builder for amqp-contract - Define type-safe AMQP messaging contracts.
pnpm add @amqp-contract/contractFor robust contract definitions with guaranteed consistency, use Event or Command patterns:
| Pattern | Use Case | Flow |
|---|---|---|
| Event | One publisher, many consumers (broadcast) | defineEventPublisher → defineEventConsumer |
| Command | Many publishers, one consumer (task queue) | defineCommandConsumer → defineCommandPublisher |
| RPC | Request / response with typed reply | defineRpc (single bidirectional definition) |
import {
defineEventPublisher,
defineEventConsumer,
defineCommandConsumer,
defineCommandPublisher,
defineContract,
defineExchange,
defineQueue,
defineMessage,
} from "@amqp-contract/contract";
import { z } from "zod";
// Event pattern: publisher broadcasts, consumers subscribe
const ordersExchange = defineExchange("orders");
const orderMessage = defineMessage(
z.object({
orderId: z.string(),
amount: z.number(),
}),
);
// Define event publisher
const orderCreatedEvent = defineEventPublisher(ordersExchange, orderMessage, {
routingKey: "order.created",
});
// Multiple queues can consume the same event
const orderQueue = defineQueue("order-processing");
const analyticsQueue = defineQueue("analytics");
// Compose contract - exchanges, queues, bindings auto-extracted
const contract = defineContract({
publishers: {
// EventPublisherConfig → auto-extracted to publisher
orderCreated: orderCreatedEvent,
},
consumers: {
// EventConsumerResult → auto-extracted to consumer + binding
processOrder: defineEventConsumer(orderCreatedEvent, orderQueue),
// For topic exchanges, consumers can override with their own pattern
trackOrders: defineEventConsumer(orderCreatedEvent, analyticsQueue, {
routingKey: "order.*", // Subscribe to all order events
}),
},
});Use defineRpc for typed request/response calls. RPC is bidirectional on both
ends — the worker handler consumes the request and produces a typed response;
the client awaits it via client.call(name, request, { timeoutMs }). Both
ends share the same definition, and RPCs live in their own rpcs slot of the
contract (not publishers or consumers). RabbitMQ direct reply-to is used
under the hood, so no reply queue declaration is needed.
import { defineContract, defineMessage, defineQueue, defineRpc } from "@amqp-contract/contract";
import { z } from "zod";
const calculate = defineRpc(defineQueue("rpc.calculate"), {
request: defineMessage(z.object({ a: z.number(), b: z.number() })),
response: defineMessage(z.object({ sum: z.number() })),
});
const contract = defineContract({
rpcs: { calculate },
});
// Server handler returns the response value, not void:
// handlers: { calculate: ({ payload }) => OkAsync({ sum: payload.a + payload.b }) }
//
// Client invokes with a required timeout:
// const result = await client.call("calculate", { a: 1, b: 2 }, { timeoutMs: 5_000 });Benefits:
- ✅ Guaranteed message schema consistency between publishers and consumers
- ✅ Routing key validation and type safety
- ✅ Full type safety with TypeScript inference
- ✅ Event, command, and RPC patterns
- ✅ Flexible routing key patterns for topic exchanges
📖 Read the full documentation →
MIT