Settlement

A handler's answer settles its AMQP delivery with a protocol frame: ack sends basic.ack, retry sends basic.reject with requeue, and drop sends basic.reject without it.

The frame goes out on the channel the delivery arrived on, and the service publishes nothing. A requeued message goes back to its queue, which hands it out again with the redelivered flag set. The next delivery may reach this consumer or another consumer of the same queue. A handler reads the flag through the Redelivered key of the delivery context. A dropped message leaves the queue: RabbitMQ discards it, or routes it to the queue's dead-letter exchange.

The crate settles with basic.reject and never with basic.nack. On one delivery the two frames do the same thing, but RabbitMQ counts a rejected delivery as spent and does not count a nacked one. So a handler that keeps asking for its message back runs down a quorum queue's delivery limit. A quorum queue stamps a returned message with the x-delivery-count header; a classic queue keeps no count, only the flag.

A retry under a cap and a delayed retry travel as copies instead of a requeue.

Example

An order is requeued the first time it arrives and acknowledged when it comes back redelivered. An order without a SKU is dropped.

Macros
Manual

examples/lapin_settlement_outcomes.rs
/// An order without a SKU is rejected for good. Any other order is requeued once, the first time
/// it arrives, and acknowledged when the queue hands it out again marked redelivered.
#[subscriber(RabbitQueue::new("orders"))]
async fn reserve(order: &Order, Ctx(redelivered): Ctx<Redelivered>) -> HandlerOutcome {
    if order.sku.is_empty() {
        tracing::warn!(order.id, "order without a SKU, rejected");
        return HandlerOutcome::drop();
    }
    if !redelivered {
        tracing::info!(order.id, "stock not counted yet, back to the queue");
        return HandlerOutcome::retry();
    }
    tracing::info!(order.id, "reserved");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("stock", "0.1.0")).with_broker(
        LapinBroker::new("amqp://localhost:5672"),
        |b| {
            b.include(reserve);
        },
    )
}

examples/manual/lapin_settlement_outcomes.rs
/// An order without a SKU is rejected for good. Any other order is requeued once, the first time
/// it arrives, and acknowledged when the queue hands it out again marked redelivered.
struct Reserve;

impl Handle<Order, (), (), AmqpContext> for Reserve {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        ctx: &mut Context<'_, AmqpContext>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        let outcome = if order.sku.is_empty() {
            tracing::warn!(order.id, "order without a SKU, rejected");
            Err(HandlerOutcome::drop())
        } else if !ctx.context(Redelivered) {
            tracing::info!(order.id, "stock not counted yet, back to the queue");
            Err(HandlerOutcome::retry())
        } else {
            tracing::info!(order.id, "reserved");
            Ok(())
        };
        ready(outcome)
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("stock", "0.1.0")).with_broker(
        LapinBroker::new("amqp://localhost:5672"),
        |b| {
            b.include(subscriber(RabbitQueue::new("orders"), Reserve).build());
        },
    )
}

Example​

Example