AMQP broker

AmqpBroker connects a service to an AMQP 1.0 broker; new only records the URL, and the runtime connects it once at startup.

AMQP 1.0 is an ISO standard, so one broker type serves ActiveMQ Artemis and Classic, RabbitMQ 4.x, Azure Service Bus, Amazon MQ, Solace, Apache Qpid and IBM MQ. AmqpBroker::new takes the broker's URL: amqp://host:port, or amqps://host:port for TLS. A URL without a port uses 5672 for amqp and 5671 for amqps. The constructor does no I/O, so the whole service fits the synchronous #[ruststream::app] builder. The settings that follow it are builder steps too: the authentication and the container id.

At startup the runtime connects the broker and receives a ConnectedAmqpBroker. The connect opens the socket, authenticates and opens the AMQP connection, so a broker that is unreachable or refuses the service fails startup instead of the first publish. Every subscription and every publisher of the service runs on that one connection. At shutdown the runtime closes the connection, and the connected form is consumed with it.

#[subscriber("orders")] opens the address orders as it is written, without a terminus capability. The broker's own configuration then decides whether the subscriptions on the address compete or each get a copy. AmqpAddress::queue and AmqpAddress::topic state that choice in the service; the Subscriptions pages describe them.

Getting started runs a first service against a local ActiveMQ Artemis.

Example

The service reads the queue address orders. cargo run --example amqp_service --features testing -- run starts it against a broker on localhost:5672 with the user artemis.

Macros
Manual

examples/amqp_service.rs
use ruststream_amqp::prelude::*;

#[subscriber(AmqpAddress::queue("orders"))]
async fn place(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "order placed");
    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(place);
        },
    )
}

examples/manual/amqp_service.rs
use ruststream_amqp::prelude::*;

struct Place;

impl Handle<Order> for Place {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(order.id, "order placed");
        ready(Ok(()))
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        AmqpBroker::new("amqp://artemis:artemis@localhost:5672"),
        |b| {
            let place = subscriber(AmqpAddress::queue("orders"), Place);
            b.include(place.build());
        },
    )
}

Example​

Example