RabbitMQ broker¶

ruststream-lapin is the RabbitMQ / AMQP 0.9.1 broker for the RustStream framework, backed by lapin. AMQP fits the framework's settlement contract natively: acks, requeues, and dead-lettering are protocol frames rather than client-side republishing. An in-process test broker ships under the testing feature.

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

LapinBroker::new is synchronous and does no I/O, so a RabbitMQ service is assembled with the same #[ruststream::app] macro as any other broker. The runtime connects the broker once at startup, before opening subscriptions; connecting consumes the broker and yields ConnectedLapinBroker, the only value carrying a subscribe or publish surface.

crates/ruststream-lapin/examples/lapin_quickstart.rs

use ruststream::runtime::{App, AppInfo, HandlerResult, RustStream};
use ruststream::{nonzero, subscriber};
use ruststream_lapin::LapinBroker;
use serde::Deserialize;

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

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

crates/ruststream-lapin/examples/lapin_quickstart.rs

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        LapinBroker::new("amqp://localhost:5672").prefetch(nonzero!(64)),
        |b| {
            b.include(handle);
        },
    )
}

The transport model¶

A subscription consumes one queue; the bare-string form #[subscriber("orders")] consumes the queue named orders, and the RabbitQueue descriptor adds bindings, queue types, and prefetch.
On the publish side the message name is the routing key; the exchange is a property of the publish policy (the default exchange unless configured). See Publishing.
Settlement is native: ack sends basic.ack, retry sends basic.nack(requeue = true), drop sends basic.reject(requeue = false) - which dead-letters when the queue has a dead-letter exchange.
Nothing is declared on the broker unless the service opts in with .declare_topology(true): infrastructure stays the user's job.

Capabilities¶

Which of the framework's optional capability traits this broker implements natively:

Capability	Native	Notes
`Subscribe`	yes	Consumes the queue the subscription names; `RabbitQueue` adds bindings, queue type, and prefetch.
`BatchSubscriber`	no	AMQP pushes one `basic.deliver` at a time, so there is no wire-level batch. Prefetch is the flow-control window instead.
`TransactionalPublisher`	yes	Both transactional publishers: `.confirms()` buffers client-side and awaits every confirm on commit, `.server_tx()` uses AMQP channel transactions. See Three publishers.
`OwnedTransactions`	yes (confirms only)	A confirms transaction is a client-side buffer, so any number can be open on one handle. `server_tx` puts the channel itself into transactional mode, which is channel state with exactly one instance.
`RequestReply`	yes	`LapinRequest` pairs into a requester over direct reply-to with correlation-id multiplexing. See Request/reply.
`Partitioned`	yes	The key travels in the `amqp-partition-key` header and feeds the runtime's worker lanes; AMQP does not interpret it, so the producer sets it. See Keyed worker lanes.
`Seekable` + `Positioned`	no	An AMQP queue is destructive: a delivery is removed from the queue when it is acked, so there is no retained history to reposition into.
`DescribeServer`	yes	Reports the configured AMQP address, which is what the AsyncAPI document records.

Scaffold a service¶

Generate a runnable starter with cargo generate, one template per messaging shape:

cargo generate --git https://github.com/powersemmi/ruststream-lapin templates/amqp-queue
cargo generate --git https://github.com/powersemmi/ruststream-lapin templates/amqp-topic

Guides¶

Queues and topology - descriptors, queue types, bindings, prefetch, dead-letter, opt-in declaration.
Publishing - the routing model, persistence, publisher confirms, and server transactions.
Request/reply - RPC over RabbitMQ direct reply-to.
Testing - the in-process test broker and the conformance harness.