Crate ruststream_amqp 

AMQP 1.0 broker for the RustStream messaging framework, backed by fe2o3-amqp.

AMQP 1.0 is an ISO standard, so one crate serves the whole family: ActiveMQ Artemis and Classic, RabbitMQ 4.x, Azure Service Bus and Event Hubs, Amazon MQ, Solace, Apache Qpid and IBM MQ. A subscription names a queue, a topic or a verbatim address, and a handler’s outcome becomes the delivery’s disposition. On RabbitMQ the 1.0 stack is a protocol of its own, separate from the 0.9.1 that ruststream-lapin speaks.

A service

A handler is an async fn over the decoded payload, mounted on a broker inside the application object, and the attribute writes the main:

use ruststream_amqp::prelude::*;
use serde::Deserialize;

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

#[subscriber(AmqpAddress::queue("orders"))]
async fn handle(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "accepted");
    HandlerOutcome::ack()
}

#[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);
        },
    )
}

cargo run -- run starts it. AmqpBroker::new only records the URL; the runtime connects the broker at startup and closes the connection when the service stops.

Where things are

The guides take an AMQP 1.0 service from installation to production, in this order: getting started, queues and topics, acknowledgement and delivery guarantees, retries and dead letters, request-reply, transactions, testing and connecting to production. The architecture pages show how AMQP 1.0 works and where the crate sits between the broker and the runtime, and the benchmarks what it costs.

Each topic below links to its component page on the site, which shows it in a service; the item’s own documentation keeps the detail.

Broker

• AMQP broker: AmqpBroker, connected into ConnectedAmqpBroker.
• container_id: AmqpBroker::container_id, the name an operator finds the service’s links by.
• Security: Sasl through AmqpBroker::sasl, and TLS for amqps:// with the rustls or native-tls feature.
• Startup: the connect that opens the socket, authenticates and begins the publishers’ session.
• Shutdown: the subscriptions, the publisher links and the connection closed in turn, and AmqpError::NotConnected for a publisher used afterwards.
• Prelude: prelude, the one glob an AMQP service imports.
• Errors: AmqpError.

Subscriptions

• Subscriptions: AmqpAddress, the address descriptor, and a bare name in #[subscriber].
• queue: AmqpAddress::queue, competing consumers, one delivery each.
• topic: AmqpAddress::topic, a copy for every subscription.
• raw: AmqpAddress::raw, an address sent as written.
• credit: AmqpAddress::credit, the window of deliveries the broker may send ahead.
• Batches: a slice handler over batches assembled on the client, closed by AmqpAddress::batch_wait.

Dispositions

• Dispositions: what ack, retry and drop send for an AmqpMessage.
• Settle: Settle, at least once or at most once.
• Retries: max_attempts, counted by the broker’s delivery-count.
• Dead letters: dead_letter, the address a spent delivery goes to.
• retry_after: a copy published to the subscription’s own address after the delay.
• out_retry: the publisher of those copies, and the steps that apply to them.

Publishing

• Publish: AmqpPublish, the publish policy, and its live AmqpPublisher.
• Headers: the well-known headers in the properties section, the rest in application-properties.
• Keys: PARTITION_KEY_HEADER rides the group-id property.
• Request-reply: request over a dynamic reply address, and responders that answer on it.
• Transactions: AmqpTransactionalPublish and its live AmqpTxnPublisher (feature transaction).

Tooling

• AMQP bindings: what the generated AsyncAPI document says about the server, the channels, the operations and the replies (feature asyncapi).
• Testing: the in-process transport TestApp connects the broker to (feature testing), and Live runs against a running broker.

Cargo features

default is empty. rustls and native-tls forward the client’s TLS backends for amqps://. transaction adds the transactional publisher. asyncapi contributes the amqp1 server and the crate’s extension to the generated document, and testing adds the in-process mode TestApp::start connects the broker through.

Modules

prelude

The imports a service on AMQP 1.0 writes every time, in one glob.

Structs

AmqpAddress

A subscription descriptor for an AMQP 1.0 address.

AmqpBroker

An AMQP 1.0 broker for the RustStream messaging framework.

AmqpMessage

A message delivered by an AmqpSubscriber.

AmqpPublish

The publish policy for AmqpPublisher: pure declaration, constructible anywhere, paired with the connected broker by the runtime after connect.

AmqpPublisher

Publishes messages to AMQP addresses, one sender link per address, attached lazily on the shared publisher session.

AmqpSubscriber

A subscription to one AMQP address; yields AmqpMessages one at a time, or in batches.

AmqpTransactionalPublish

The publish policy for AmqpTxnPublisher: names the transactional publishing mode as a distinct policy type, so the plain publisher carries no transactional surface at all.

AmqpTxnPublisher

A publisher whose messages between begin_transaction and commit become visible atomically, over the protocol’s transactional state.

ConnectedAmqpBroker

The typed witness that connect succeeded: holds the live connection directly.

Sasl

How the broker authenticates the connection, mapped onto the client’s SASL profiles.

Enums

AmqpError

Errors returned by the AMQP 1.0 broker.

Settle

Delivery guarantee of a subscription, set with AmqpAddress::settle.

Constants

DEFAULT_BATCH_WAIT

Default deadline closing a partial batch on a batch subscription.

DEFAULT_CREDIT

Default link credit granted to a subscription: the window of deliveries the broker may send ahead of the handler’s settlements.

PARTITION_KEY_HEADER

Header carrying the partition key, mapped onto the AMQP group-id property.