delay

delay(Delay::dlx_ttl()) makes retry_after wait on the broker: the copy parks in a waiting queue under a per-message TTL.

On a retry_after, the crate publishes the copy to the waiting queue with the delay as its expiration, then acknowledges the original. Both frames go out on one channel in that order, so a connection lost between them brings the original back rather than losing it. The waiting queue has no consumers. When the TTL fires, it dead-letters the copy back to the origin queue on the default exchange. The copy lives on the broker, so it survives a restart of the service. The waiting queue is <queue>.retry; Delay::dlx_ttl_named names another. The released copy is a new message to the server, and it carries x-ruststream-retry-count raised by one. So a retry cap ends a delayed loop the way it ends an immediate one.

A waiting queue releases expired messages only from its head, so a short delay waits behind a long one. Mixed delays take one waiting queue per delay class, or the delayed-message plugin.

The waiting queue is infrastructure. With declare_topology(true) the service declares it: durable, dead-lettering to the origin queue. On a broker that declares nothing, the operator provisions it with those arguments. The subscription checks at startup that the waiting queue exists, and the service fails to start without it.

Example

A shipment that arrives while the warehouse is closed waits a minute in shipments.retry. The released copy carries the retry count, and the shipment goes out.

Macros
Manual

examples/lapin_settlement_delay.rs
/// A shipment that arrives while the warehouse is closed waits a minute in `shipments.retry`, on
/// the broker. The copy it releases carries the retry count, and the shipment goes out.
#[subscriber(RabbitQueue::new("shipments").delay(Delay::dlx_ttl()))]
async fn dispatch(shipment: &Shipment, ctx: &mut Context<'_>) -> HandlerOutcome {
    if ctx.headers().get_str(RETRY_COUNT_HEADER).is_none() {
        tracing::info!(shipment.id, "warehouse closed, waiting");
        return HandlerOutcome::retry_after(WAIT);
    }
    tracing::info!(shipment.id, "dispatched");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    // The service declares its queues, `shipments.retry` included.
    let broker = LapinBroker::new("amqp://localhost:5672").declare_topology(true);
    RustStream::new(AppInfo::new("shipping", "0.1.0")).with_broker(broker, |b| {
        b.include(dispatch);
    })
}

examples/manual/lapin_settlement_delay.rs
/// A shipment that arrives while the warehouse is closed waits a minute in `shipments.retry`, on
/// the broker. The copy it releases carries the retry count, and the shipment goes out.
struct Dispatch;

impl Handle<Shipment> for Dispatch {
    fn handle(
        &self,
        shipment: &Shipment,
        _outs: &(),
        ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        if ctx.headers().get_str(RETRY_COUNT_HEADER).is_none() {
            tracing::info!(shipment.id, "warehouse closed, waiting");
            return ready(Err(HandlerOutcome::retry_after(WAIT)));
        }
        tracing::info!(shipment.id, "dispatched");
        ready(Ok(()))
    }
}

fn app() -> impl App {
    // The service declares its queues, `shipments.retry` included.
    let broker = LapinBroker::new("amqp://localhost:5672").declare_topology(true);
    RustStream::new(AppInfo::new("shipping", "0.1.0")).with_broker(broker, |b| {
        let shipments = RabbitQueue::new("shipments").delay(Delay::dlx_ttl());
        b.include(subscriber(shipments, Dispatch).build());
    })
}

Example​

Example