Publish messages

Recipes for the publishing side. For the concepts behind the result type, see errors as values.

Create a client

import { TypedAmqpClient } from "@amqp-contract/client";

import { contract } from "./contract.js";

const client = await TypedAmqpClient.create({
  contract,
  urls: ["amqp://localhost"],
}).getOrThrow();

create returns AsyncResult<TypedAmqpClient, ConnectionError>. The .getOrThrow() is required to reach the client — awaiting alone leaves you with a Result. An unreachable broker is modeled rather than a defect, so a start-up path can triage it by tag instead of recovering every defect to get at it; .getOrThrow() says you would rather throw.

To log that failure before it propagates:

const client = await TypedAmqpClient.create({ contract, urls: ["amqp://localhost"] })
  .tapFailure((failure) =>
    logger.error(
      { error: failure.tag === "Err" ? failure.error : failure.cause },
      "could not connect to the broker",
    ),
  )
  .getOrThrow();

tapFailure observes both channels without consuming — the modeled ConnectionError and a defect from an outright bug — so .getOrThrow() still throws afterwards. To handle instead of throwing, .match(...) on the tag.

Pass several URLs for failover — the client tries them in order:

urls: ["amqp://primary:5672", "amqp://secondary:5672"];

The client declares what its publishes need to be routed and retained: its publishers' exchanges, the queues they route to (with their bindings) and the RPC request queues — so a message published before the worker starts waits in its queue. It declares their dead-letter exchanges and DLQs too, so a message that expires or overflows before any worker runs is dead-lettered rather than discarded, and leaves the retry wait queues to the worker. Set topology: "passive" to only check that they exist (for credentials that may not configure the broker), or topology: "none" when topology is provisioned elsewhere.

For a readiness probe, client.isConnected() reports whether the broker connection is up; it reads false while the client is reconnecting.

Publish a message

const result = await client.publish("orderCreated", {
  orderId: "ORD-123",
  customerId: "CUST-456",
  amount: 99.99,
});

The first argument must be a publisher name from the contract; the second is checked against that publisher's schema. Both are compile errors when wrong.

Handle the outcome:

import { P } from "unthrown";

result.match({
  ok: () => console.log("published"),
  errCases: (matcher) =>
    matcher
      .with(P.tag("@amqp-contract/MessageValidationError"), (error) =>
        console.error("invalid payload:", error.issues),
      )
      .with(P.tag("@amqp-contract/PublishError"), (error) =>
        console.error(`broker side failed (${error.reason}):`, error.message),
      ),
  defect: (cause) => {
    throw cause; // a bug, not an operational condition
  },
});

publish returns AsyncResult<void, MessageValidationError | PublishError>. PublishError is the broker side failing, with a reason: "timeout" (buffered past publishTimeoutMs while the broker was unreachable), "nacked" (the broker refused the message) or "channel-closed". A full write buffer is not a failure: the message was confirmed before the backpressure was reported, so publish answers Ok. Buffer, retry, shed load or surface a 503 — it is yours to handle. Anything else (an unencodable payload, a rejection nobody anticipated) arrives as a defect. Each error class exposes its tag as a static, so P.tag(PublishError.tag) works too.

Set default options for every publish

const client = await TypedAmqpClient.create({
  contract,
  urls: ["amqp://localhost"],
  defaultPublishOptions: {
    priority: 5,
    headers: { "x-app-version": "1.0.0" },
  },
}).getOrThrow();

Per-call options override these. Messages are persistent by default; set persistent: false here or per call to opt out.

Send a message on a different routing key

You cannot. The routing key comes from the publisher definition and there is no per-call override — publish uses publisher.routingKey and nothing else.

To publish the same payload on several keys, define a publisher per key and let them share a message:

const orderCreated = defineEventPublisher(ordersExchange, orderMessage, {
  routingKey: "order.created",
});
const orderCreatedUrgent = defineEventPublisher(ordersExchange, orderMessage, {
  routingKey: "order.created.urgent",
});

await client.publish(urgent ? "orderCreatedUrgent" : "orderCreated", order);

This is more verbose than a per-call key, deliberately: every routing key a service can emit stays visible in the contract, and therefore in the generated AsyncAPI document.

Send headers

await client.publish(
  "orderCreated",
  { orderId: "ORD-123", amount: 99.99 },
  {
    headers: {
      eventSource: "checkout-service",
      eventVersion: 2,
    },
  },
);

Headers are not validated on publish — only the payload is. If the message declares a headers schema, it is enforced by the consumer, and a mismatch dead-letters the message there rather than failing the publish here. Treat the headers schema as part of the contract and keep the two in step yourself.

Set AMQP properties

PublishOptions is amqplib's Options.Publish plus compression, so properties are set flat, alongside headers:

await client.publish(
  "orderCreated",
  { orderId: "ORD-123", amount: 99.99 },
  {
    persistent: false,
    priority: 10,
    expiration: "60000",
    correlationId: "req-123",
    headers: { "x-request-id": "req-123" },
  },
);

Fail the process on a publish error

In a script or a job where a failed publish should stop everything:

await client.publish("orderCreated", order).getOrThrow();

getOrThrow throws the MessageValidationError or PublishError on Err and rethrows a defect's cause. Prefer .match in long-running services, where you usually want to log and continue rather than take the process down.

Log failures without handling them

To observe a failure while leaving it in the pipeline:

await client
  .publish("orderCreated", order)
  .tapFailure((failure) =>
    logger.error(
      { error: failure.tag === "Err" ? failure.error : failure.cause },
      "publish failed",
    ),
  )
  .tap(() => logger.debug("published"))
  .getOrThrow();

tapFailure sees both channels — failure.tag discriminates Err from Defect. tap only fires on success.

Publish many messages

There is no batch API; publish in a loop and collect the results.

import { allAsync } from "unthrown";

const results = await allAsync(orders.map((order) => client.publish("orderCreated", order)));

allAsync fails on the first error. To publish everything regardless and report afterwards, await them individually and partition the results yourself. For throughput considerations see tune performance.

Close the client

await client.close().get();

The .get() matters — without it the close result is discarded and a failure passes unnoticed. Closing fails any in-flight RPC calls with RpcCancelledError.

In a service, close on shutdown:

process.on("SIGTERM", async () => {
  await client.close().get();
  process.exit(0);
});

Where next

• Use request/reply — when you need an answer back.
• Add middleware — stamp headers on every publish.
• Compress messages — for large payloads.
• Error model — every error publish and call can produce.