RabbitMQ broker

ruststream-lapin runs a RustStream service on RabbitMQ over lapin, the AMQP 0.9.1 client. A RabbitMQ queue is destructive storage: a delivery leaves the queue once it is acknowledged. Acknowledgement, requeue and dead-lettering are protocol frames, so the framework settles a message on the broker itself.

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

LapinBroker::new 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 at startup, before it opens subscriptions. Connecting consumes the broker and yields ConnectedLapinBroker, the only value carrying a subscribe or publish surface.

A handler body names capabilities: Out<impl Publisher>, Out<impl TransactionalPublisher>, Out<impl RequestReply>. It imports the framework's prelude alone, which is what keeps one handler mountable under whichever publish policy the routes file pairs it with. One body imports this crate's prelude too: one that adjusts an AMQP property for a single message. It names the settings type rather than a publisher type - Out<impl Publisher<Options = LapinPublishOptions>> - so the handler still names no publisher.

A routes file names values and imports ruststream_lapin::prelude::*, which brings the framework's prelude with it. It adds the family's uniform mount-site names: Publish, TransactionalPublish and Request, aliases for LapinPublish, ConfirmsPublish and LapinRequest.

crates/ruststream-lapin/examples/lapin_quickstart.rs

use ruststream::nonzero;
use ruststream_lapin::prelude::*;
use serde::Deserialize;

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

#[subscriber("orders")]
async fn handle(order: &Order) -> HandlerOutcome {
    println!("got order {}", order.id);
    HandlerOutcome::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. #[subscriber("orders")] consumes the queue named orders, and the RabbitQueue descriptor adds bindings, queue types and prefetch. A batch handler names its size at the mount site, and the crate assembles the batch on the client, since AMQP has no wire batch; see Batches.
• On the publish side the message name is the routing key, and the exchange belongs to the publish policy: the default exchange unless you name another. See Publishing.
• Settlement is native: ack sends basic.ack, retry sends basic.reject(requeue = true), drop sends basic.reject(requeue = false), which dead-letters the message when the queue has a dead-letter exchange. A rejection is the frame RabbitMQ counts as a spent delivery, so a handler's retry runs down a quorum queue's own limit.
• Nothing is declared on the broker unless you opt in with .declare_topology(true): the topology is yours to manage.

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	yes, client-side	AMQP pushes one basic.deliver at a time, so the batch is assembled on the client, to the size the mount site names, under the descriptor's prefetch window and batch_wait. See Batches.
TransactionalPublisher	yes	Both transactional publishers: .confirms() buffers client-side and awaits every confirm on commit, .server_tx() uses AMQP channel transactions. See Publishing.
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 producer sets the key in the amqp-partition-key header and the runtime's worker lanes read it; AMQP itself does not interpret it. See Delivery metadata.
Seekable + Positioned	no	A queue keeps no history to reposition into.
DescribeServer	yes	Reports the connection host and the AMQP version behind it, which is what the AsyncAPI document records. See The AsyncAPI document.

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

Documentation

The crate documents itself on docs.rs, and that overview is the guide:

• Subscribing - the two queue descriptors, queue types, bindings, prefetch, the retry cap, delayed redelivery, batches and delivery metadata.
• Publishing - the routing model, the policies, the per-message AMQP properties, replies and transactions.
• Request/reply - RPC over direct reply-to.
• The AsyncAPI document - the AMQP bindings a service reports about itself.
• Testing - the production app under the TestApp harness, in process or against a live broker.
• Operations - TLS, connection settings, opt-in declaration, known gaps.

Installation, the tutorial and the other brokers are on the RustStream site: https://powersemmi.github.io/ruststream/.