Publishing

The message name is the routing key. The exchange is a property of the publisher: the default exchange unless .exchange("events") says otherwise. On the default exchange the routing key addresses the queue with that name, which is why the quickstart works with no topology at all.

Messages are published persistent (delivery mode 2) by default; .persistent(false) opts out for fire-and-forget traffic where losing messages on a broker restart is acceptable.

Well-known headers map onto native AMQP properties (content-type, correlation-id, reply-to, message-id); every other header travels in the AMQP header table as a byte string, so binary values round-trip.

Replying from a handler

The framework's publish(..) form works unchanged: the handler returns the reply value and the runtime encodes and publishes it through the TypedPublisher the mount was given (see the core publishing guide for the whole surface, including per-publisher transforms and app-wide publish layers). The request/reply page shows the RPC variant, where a transform redirects each reply to the requester's private address.

Three publishers

broker.publisher() is fire-and-forget: the publish resolves when the frame is written, with no broker feedback. Upgrade on the publisher when the guarantee matters:

• .confirms() - publisher confirms: every publish resolves only once the broker confirmed it. Transactions buffer client-side and flush on commit. Durable and fast; the recommended transactional publisher.
• .server_tx() - AMQP channel transactions (tx.select / tx.commit / tx.rollback): messages become visible atomically at commit. Slower (a synchronous round trip per commit), but the only option when partial flushes are unacceptable.

// The transactional flavour is picked on the publisher; swap `.confirms()` for
// `.server_tx()` to trade throughput for AMQP server-side atomicity.
let shipments = Shipments {
    publisher: broker.publisher().confirms(),
};

The trade-off in one sentence: confirms give per-message durability (a failed commit may leave earlier messages published), server transactions give all-or-nothing visibility.

Transactional fan-out from a handler

A publisher is a value like any other shared resource: build it once, wire it into the typed application state at startup, and let handlers request it by type (State<Shipments> below). Here an order fans out into per-item shipment commands, published all-or-nothing:

// The application state wires the use-case object once at startup; `#[derive(FromRef)]` makes
// each field injectable into handlers as `State<FieldType>`.
#[derive(Clone, FromRef)]
struct AppState {
    shipments: Shipments,
}

#[derive(Clone)]
struct Shipments {
    publisher: ConfirmsPublisher,
}

impl Shipments {
    /// Publishes one shipment command per item, all-or-nothing: commit resolves only after the
    /// broker confirmed every message, and any failure aborts so shipments are never
    /// half-visible.
    async fn dispatch(&self, order: &Order) -> Result<(), AmqpError> {
        self.publisher.begin_transaction().await?;
        for item in &order.items {
            let command = ItemShipment {
                order_id: order.id,
                item: item.clone(),
            };
            let payload = JsonCodec.encode(&command).expect("serializable");
            let outgoing = OutgoingMessage::new("shipments", payload.as_ref());
            if let Err(err) = self.publisher.publish(outgoing).await {
                self.publisher.abort().await.ok();
                return Err(err);
            }
        }
        self.publisher.commit().await
    }
}

#[subscriber("orders")]
async fn ship(order: &Order, State(shipments): State<Shipments>) -> HandlerResult {
    if shipments.dispatch(order).await.is_err() {
        // Nothing was committed; ask for redelivery and try the whole fan-out again.
        return HandlerResult::retry();
    }
    HandlerResult::Ack
}

Both transactional publishers implement the framework's TransactionalPublisher, so either plugs into the same begin_transaction / commit / abort call sites. Clones of a publisher share the underlying channel and transaction state.