ruststream-lapin 0.7.1

ruststream-lapin connects a RustStream service to RabbitMQ over lapin, the AMQP 0.9.1 client. Handlers, routing, codecs and middleware come from the framework; this crate is the transport.

Features

• Native settlement: ack is basic.ack, a retry is basic.reject with requeue, and a drop goes to the queue's dead-letter exchange.
• A descriptor per queue type: RabbitQueue for classic queues, RabbitQuorumQueue for quorum queues, with bindings, prefetch and dead-letter settings.
• Your topology: nothing is created on the broker unless the service opts in with declare_topology(true).
• Durable delayed retry through a TTL waiting queue or the delayed-message-exchange plugin, and a retry cap a quorum queue enforces itself.
• Three publishers: fire-and-forget, publisher confirms with client-side transactions, and AMQP channel transactions. Priority, TTL and persistence are typed per-message settings.
• Request/reply over direct reply-to.
• AsyncAPI with the specification's amqp bindings, behind the asyncapi feature.
• Tests without a server: the service's own app runs with LapinBroker in process.

Install

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

[dev-dependencies]
ruststream-lapin = { version = "0.7", features = ["testing"] }

Optional features: asyncapi, TLS (tls-rustls, tls-rustls-ring, tls-native-tls), and the broker plugins plugin-consistent-hash and plugin-dme.

Write a service

use ruststream_lapin::prelude::*;
use serde::{Deserialize, Serialize};

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

#[derive(Debug, Deserialize, Outgoing, PartialEq, Serialize)]
#[outgoing(name = "receipts")]
struct Receipt {
    order_id: u64,
}

#[subscriber(RabbitQueue::new("orders").durable(true))]
async fn settle(order: &Order, Out(receipts): Out<impl Publisher>) -> HandlerOutcome {
    let receipt = Receipt { order_id: order.id };
    if receipts.message(&receipt).publish().await.is_err() {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    let broker = LapinBroker::new("amqp://localhost:5672").prefetch(nonzero!(64));
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(settle).out(DefaultSlot, Publish::default()).build();
    })
}

#[ruststream::app] generates main, so the binary understands run and asyncapi gen.

Scaffold a fresh project from a template, a work queue or a topic exchange:

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

Test it

TestApp runs the service's own app with LapinBroker in process, with no server.

use ruststream::testing::TestApp;

let tb = TestApp::start(app()).await?;

tb.broker::<LapinBroker>()
    .message(&Order { id: 42 })
    .to("orders")
    .publish()
    .await?;

tb.broker::<LapinBroker>()
    .subscriber("orders")
    .assert_called_once()
    .settled(HandlerOutcome::ack());

tb.broker::<LapinBroker>()
    .published::<Receipt>("receipts")
    .assert_called_once()
    .with(&Receipt { order_id: 42 });

TestApp::start_live(app()) runs the same test against a running RabbitMQ.

Documentation

• Getting started - installation, the transport model and the service scaffolds.
• Guides - building a RabbitMQ service with the crate.
• Components - one page per part of the crate.
• Architecture - how the crate sits between lapin and the runtime.
• Benchmarks - what the crate costs over the raw client.
• API reference - every item on docs.rs; the crate overview maps each topic to its page.
• The framework: https://powersemmi.github.io/ruststream/latest/

Minimum supported Rust version

The MSRV is 1.95, edition 2024.

Contributing

See CONTRIBUTING.md.

License

Licensed under the Apache-2.0 license.