Subscriptions

A queue name in #[subscriber] consumes that existing queue as a classic queue; a descriptor carries every other setting.

One subscription consumes one queue. #[subscriber("orders")] consumes the queue orders with the defaults of RabbitQueue: a durable classic queue, the broker's prefetch and no bindings. The queue has to exist on the server, unless the broker declares the topology. A message reaches it through the default exchange, which routes on the queue's name, or through an exchange the queue is bound to.

RabbitMQ has two queue types, and the type is fixed when the queue is created. The type decides what a retry does, so each type has a descriptor of its own:

Descriptor	Queue	A retry is
`RabbitQueue`	classic, on one node	counted by the runtime, on the copies it publishes back
`RabbitQuorumQueue`	quorum, replicated	counted by the server, which ends a message that spent its deliveries

A descriptor goes where the name goes, as the attribute's first argument, and every one of its steps returns the descriptor. The manual form passes the name or the descriptor to subscriber(source, body). Several instances of a service share one queue, and each message goes to one of them. The handler's outcome is a protocol frame: an ack removes the message, and a retry or a drop rejects it, as the Settlement pages show.

Example

An order with items is acknowledged; an empty order is dropped.

Macros
Manual

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

#[subscriber("orders")]
async fn place(order: &Order) -> HandlerOutcome {
    if order.items == 0 {
        tracing::warn!(order.id, "empty order dropped");
        return HandlerOutcome::drop();
    }
    tracing::info!(order.id, order.items, "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_subscriptions_queue.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>> {
        if order.items == 0 {
            tracing::warn!(order.id, "empty order dropped");
            return ready(Err(HandlerOutcome::drop()));
        }
        tracing::info!(order.id, order.items, "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