LapinBroker

LapinBroker connects a service to RabbitMQ; new only records the URI, and the runtime connects it at startup and shuts it down at the end.

LapinBroker::new takes the connection URI. It does no I/O, so the whole service fits the synchronous #[ruststream::app] builder. The settings that follow it are builder steps too: a connection name, a default prefetch window and declare_topology.

At startup the runtime connects the broker and receives a ConnectedLapinBroker. The connect opens the AMQP connection and logs in, so an unreachable server or a refused login fails startup instead of the first delivery. Every subscription and every publisher of the service opens against that connected form. At shutdown the runtime closes the connection and receives a ClosedLapinBroker.

#[subscriber("orders")] consumes the queue orders. The queue has to exist on the server, unless the broker declares the topology. The descriptors that bind a queue to an exchange or choose its type are on the Subscriptions pages.

Example

The service consumes the queue orders on a server at localhost:5672. cargo run --example lapin_broker_service --features testing -- run starts it.

Macros
Manual

examples/lapin_broker_service.rs
use ruststream_lapin::prelude::*;

#[subscriber("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(
        LapinBroker::new("amqp://localhost:5672"),
        |b| {
            b.include(place);
        },
    )
}

examples/manual/lapin_broker_service.rs
use ruststream_lapin::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(
        LapinBroker::new("amqp://localhost:5672"),
        |b| {
            b.include(subscriber("orders", Place).build());
        },
    )
}

Example​

Example