AMQP 1.0¶

ruststream-amqp is the AMQP 1.0 broker, built on fe2o3-amqp. The protocol is an ISO standard, so the same crate talks to ActiveMQ Artemis, RabbitMQ 4.x, Azure Service Bus, Amazon MQ, Solace, Apache Qpid, and IBM MQ. For framework concepts (writing subscribers, routing, codecs, middleware), see the RustStream documentation.

ruststream = { version = "0.6", features = ["macros", "json"] }
ruststream-amqp = "0.6"
serde = { version = "1", features = ["derive"] }

Capabilities¶

The framework's optional capability traits, and what this broker implements natively:

Capability	Native	Notes
`Subscribe`	yes	subscribe by name; the name is sent verbatim, as `AmqpAddress::raw` does
`BatchSubscriber`	no	the protocol delivers one message per transfer, and batching is credit, not a batch pull
`TransactionalPublisher`	yes, with the `transaction` feature	transactional posting, one broker-side transaction per handle
`OwnedTransactions`	no	only the borrowed form is implemented; the client's transactional path covers posting
`RequestReply`	yes	`reply-to`, `correlation-id`, and a dynamic reply link
`Partitioned`	yes	the partition key rides the `group-id` property
`Seekable` and `Positioned`	no	the queue position belongs to the broker; the protocol exposes no client-addressable offset to seek to
`DescribeServer`	yes	reports the connection host and the `amqp` protocol for the framework's server description

The lifecycle¶

The broker is a ladder of consuming transitions, so each state is a distinct type:

AmqpBroker::new(url)      configuration only, synchronous, no I/O
  .connect()   ->  ConnectedAmqpBroker      the live connection; subscriptions and publishers
  .shutdown()  ->  ()                       ends the sessions and closes the connection

new performs no I/O, so an AMQP service is assembled with the same #[ruststream::app] macro as any other broker: the runtime connects once at startup, before opening subscriptions, and closes the connection at the end. Because shutdown consumes the connected broker, publishing or subscribing after it does not compile. A publisher handed out earlier still aliases the connection, and reports AmqpError::NotConnected once it is gone rather than succeeding against a dead connection.

Authentication and identity are builder options on the synchronous form: sasl takes a Sasl::anonymous(), Sasl::plain(user, pass), or Sasl::external() profile, and container_id names the service to the broker (the default is "ruststream"). A URL with userinfo (amqp://user:pass@host) selects PLAIN implicitly, and an explicit profile wins over it. TLS endpoints (amqps://) need the rustls or native-tls feature.

Each subscription runs on its own AMQP session, and publishers share one session of their own. Flow-control windows are per session, so a slow consumer cannot starve the publishers or another subscription.

Addressing¶

AMQP 1.0 standardises the wire but not the meaning of an address, so AmqpAddress makes the intent explicit. Each constructor advertises the matching terminus capability, which is how Artemis and other products disambiguate:

Constructor	Semantics	Terminus capability
`AmqpAddress::queue(name)`	anycast: competing consumers, one delivery each	`queue`
`AmqpAddress::topic(name)`	multicast: fan-out to every subscriber	`topic`
`AmqpAddress::raw(address)`	verbatim, for a deployment's own convention	none

AmqpAddress implements SubscriptionSource, so the descriptor sits inline in the decorator:

crates/ruststream-amqp/examples/amqp_service.rs

#[derive(Debug, Deserialize)]
struct Order {
    id: u64,
}

#[subscriber(AmqpAddress::queue("orders"))]
async fn handle(order: &Order) -> HandlerResult {
    println!("got order {}", order.id);
    HandlerResult::Ack
}

Wiring it onto the broker is the framework's with_broker / include pair, identical to every other broker:

crates/ruststream-amqp/examples/amqp_service.rs

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        AmqpBroker::new("amqp://artemis:artemis@localhost:5672"),
        |b| b.include(handle),
    )
}

Two options ride the descriptor:

credit(n) sets the protocol-level credit (prefetch): how many unsettled deliveries the broker may have in flight to this subscription. The default is 256. Credit is the protocol's own back-pressure, so a lower value bounds work in flight without an extra layer.
settle(Settle::AtMostOnce) switches the subscription to at-most-once delivery, where the receiver settles on receipt.

A descriptor that cannot form a subscription (an empty address, zero credit) is rejected with AmqpError::InvalidAddress before any I/O.

The plain string form #[subscriber("orders")] also works: a by-name source resolves to AmqpAddress::raw, so the address goes to the broker verbatim with no capability attached.

Acknowledgement and dispositions¶

Settlement maps onto the protocol's dispositions, with no invented middle layer:

Handler result	Disposition	Effect
`HandlerResult::Ack`	`accept`	the delivery is done, the broker drops it
`HandlerResult::retry()`	`release`	the delivery returns to the broker for redelivery
`HandlerResult::drop()`	`reject`	terminal; the broker's dead-letter policy decides

On an at-most-once subscription the deliveries arrive already settled, so ack and nack report AckError::Unsupported instead of a settlement that never reaches the wire.

AMQP 1.0 has no protocol-level delayed redelivery, so HandlerResult::retry_after(delay) falls back to the runtime's broker-agnostic deferred re-publish rather than a broker-side timer.

Publishing¶

A publisher is a policy plus the live connection. AmqpPublish holds no connection, so it is constructed anywhere (in a router, in configuration, at a mount site) and the runtime pairs it with the broker at startup to produce an AmqpPublisher. It is also the broker's default publish policy, so a #[subscriber(.., publish("dest"))] handler mounted without an explicit publisher replies through it.

Sender links are attached on first use and cached per address. A message the peer settles with anything other than accept (rejected, released, modified) is reported as AmqpError::PublishNotAccepted, so a broker-side refusal cannot pass as a successful publish.

Request/reply¶

AMQP 1.0 carries request/reply natively, so AmqpPublisher implements the RequestReply capability. request(msg, timeout) attaches a dynamic receiver link (the broker mints a private reply address), sends the message with reply-to and correlation-id set, and resolves with the first reply carrying the matching correlation id. Nothing answering within the timeout is an AmqpError::RequestTimeout, and the reply link is detached either way.

The requester side is a first publish, so it belongs in the scope's after_startup hook, where the publisher arrives live, already paired with the connected broker:

crates/ruststream-amqp/examples/amqp_request_reply.rs

b.after_startup(AmqpPublish, async move |publisher| -> io::Result<()> {
    let reply = publisher
        .request(
            OutgoingMessage::new("greeter", b"world".as_slice()),
            Duration::from_secs(5),
        )
        .await
        .map_err(io::Error::other)?;
    println!("reply: {}", String::from_utf8_lossy(reply.payload()));
    Ok(())
});

The responder end reads the reply address the requester named and publishes the answer there. The address is minted per request, so the fixed-destination publish(..) reply form does not fit: the reply rides an injected publisher, and echoes correlation-id so a late reply cannot resolve a later request.

crates/ruststream-amqp/examples/amqp_request_reply.rs

/// The responder. A reply goes to the address the requester named in `reply-to`, which the broker
/// mints per request, so the fixed-destination `publish(..)` form does not fit: the reply rides an
/// injected publisher and echoes `correlation-id` so a late reply cannot resolve a later request.
#[subscriber(AmqpAddress::queue("greeter"), raw)]
async fn greet(name: &[u8], ctx: &mut Context<'_>, Out(out): Out<AmqpPublisher>) -> HandlerResult {
    let Some(reply_to) = ctx.headers().reply_to().map(str::to_owned) else {
        return HandlerResult::drop();
    };
    let mut headers = Headers::new();
    if let Some(correlation_id) = ctx.headers().correlation_id() {
        headers.insert("correlation-id", correlation_id.to_owned());
    }

    let payload = format!("hello, {}", String::from_utf8_lossy(name));
    let reply = OutgoingMessage::new(&reply_to, payload.as_bytes()).with_headers(headers);
    if out.publish(reply).await.is_err() {
        return HandlerResult::retry();
    }
    HandlerResult::Ack
}

Both sides of the exchange are in examples/amqp_request_reply.rs.

Transactions¶

With the transaction feature, AmqpTransactionalPublish pairs into an AmqpTxnPublisher, which implements TransactionalPublisher over the protocol's transactional posting. The transactional mode is a separate policy type, so the plain publisher carries no transactional surface at all.

crates/ruststream-amqp/examples/amqp_transaction.rs

b.after_startup(
    AmqpTransactionalPublish,
    async move |publisher| -> io::Result<()> {
        publisher
            .begin_transaction()
            .await
            .map_err(io::Error::other)?;

        for id in 1..=3_u64 {
            let payload = format!("{{\"id\":{id}}}");
            let message = OutgoingMessage::new("invoices", payload.as_bytes());
            if let Err(error) = publisher.publish(message).await {
                publisher.abort().await.map_err(io::Error::other)?;
                return Err(io::Error::other(error));
            }
        }

        publisher.commit().await.map_err(io::Error::other)
    },
);

The handle carries at most one broker-side transaction: a second begin_transaction while one is open is an error that leaves the open transaction untouched, and commit or abort with nothing open is an error rather than a silent no-op. A failed discharge still closes the transaction, so the next begin_transaction starts fresh and the handle never wedges. Publishing outside a transaction goes out immediately, on the same publisher.

The scope is transactional posting only. Transactional retirement (settling deliveries inside a transaction) and transactional acquisition are not implemented.

Headers and the partition key¶

Well-known headers ride the AMQP properties section: content-type, correlation-id, reply-to, message-id, and the partition key as group-id. Every other header rides application-properties. No envelope format is invented, so a non-Rust peer sees a plain AMQP message and a message produced by another AMQP client arrives with its headers intact.

The partition key is read and written through the partition-key header (exported as PARTITION_KEY_HEADER), which is the same convention the framework's in-memory broker uses, and delivered messages implement the Partitioned capability.

Testing¶

The testing feature ships AmqpTestBroker: an in-process transport that reproduces the crate's core routing with no server and no AMQP wire. It follows the same ladder as the real broker, and its connected form implements ruststream::testing::TestableBroker, so the same broker drives the TestApp harness and the framework's conformance suite. Inject traffic with broker.inject(OutgoingMessage::new(..)) and assert on published output with the free ruststream::testing::expect_published. See Unit-testing a service with TestApp.

The test broker routes by exact address match and does not simulate broker-specific behaviour (dead-letter policies, credit, redelivery timing). Those are verified end to end against a real broker: just test-brokers starts ActiveMQ Artemis from docker-compose.test.yml and runs the integration tests plus the conformance lifecycle, request/reply, and transactions suites against it, gated behind AMQP_TEST_URL.