ruststream-amqp 0.7.1

ruststream-amqp connects a RustStream service to an AMQP 1.0 broker over fe2o3-amqp. One crate serves the whole family: ActiveMQ Artemis and Classic, RabbitMQ 4.x over AMQP 1.0, Azure Service Bus and Event Hubs, Amazon MQ, Solace, Apache Qpid and IBM MQ. For RabbitMQ over AMQP 0.9.1, use ruststream-lapin. Handlers, routing, codecs and middleware come from the framework; this crate is the transport.

Features

• Settlement as dispositions: ack is accepted, a retry is modified, a drop is rejected, so the broker's own dead-letter policy applies.
• Explicit addressing: queues (anycast), topics (multicast) or a raw address, with credit as flow control.
• Retry caps and dead letters declared where the handler is mounted.
• Batches assembled on the client.
• Request/reply over reply-to and correlation-id, and transactions behind the transaction feature.
• Plain AMQP messages for non-Rust peers: well-known headers ride the properties section.
• AsyncAPI that names the protocol amqp1, behind the asyncapi feature.
• Tests without a server: handlers run against an in-process AMQP broker.

Install

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

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

Optional features: TLS for amqps:// (rustls or native-tls), transaction and asyncapi.

Write a service

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

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

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

#[subscriber(AmqpAddress::queue("orders"))]
async fn handle(order: &Order, Out(out): Out<impl Publisher>) -> HandlerOutcome {
    let confirmation = Confirmation { order_id: order.id };
    if out.message(&confirmation).publish().await.is_err() {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

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

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

Test it

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

use ruststream::testing::TestApp;

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

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

tb.broker::<AmqpBroker>()
    .subscriber("orders")
    .assert_called_once()
    .with(&Order { id: 42 })
    .settled(HandlerOutcome::ack());

tb.broker::<AmqpBroker>()
    .published::<Confirmation>("confirmations")
    .assert_called_once()
    .with(&Confirmation { order_id: 42 });

Documentation

• Getting started - installation and a first service against a local ActiveMQ Artemis.
• Guides - queues and topics, acknowledgement, retries and dead letters, request-reply, transactions, testing and a production broker.
• Components - one page per part of the crate: the broker, subscriptions, dispositions, publishing, request-reply, transactions, the AsyncAPI bindings and testing.
• Architecture - how AMQP 1.0 works and how the crate sits between fe2o3-amqp and the runtime.
• Benchmarks - what the crate costs over the raw client.
• API reference - every item on docs.rs.
• 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.