Skip to main content

retry_after

On a stream file a retry_after copy is appended under the key the subscription reads, so it comes back with nothing named at the mount site.

retry_after(delay) asks for a later delivery. A stream file cannot hold a message back, so the runtime does it. It settles the delivery, waits out the delay and publishes a copy of the message, payload and headers, through the form's Publish. One stream key is both ends of the file, so the copy is appended under the key the subscription reads, and the subscription reads it as the next message. This holds for a FileStream and for a bare stream key alike. The copy carries RETRY_COUNT_HEADER incremented, and the handler reads it like any other header. The copy waits in the service's memory, so a process that exits before the delay is over loses it. A replay reads the copy while the replay is still short of the end of the file, and leaves it unread once it has passed that point. out_retry replaces the publisher the copy leaves through, and a transform on that position runs on the copy.

Example​

The first attempt finds the card processor busy; the copy comes back 30 seconds later and is charged.

examples/file_retry_after.rs
/// Charges a payment; the first attempt finds the card processor busy and asks again later.
#[subscriber(FileStream::new("payments"))]
async fn charge(payment: &Payment, ctx: &mut Context<'_>) -> HandlerOutcome {
    // A copy carries the framework's retry count; the first delivery carries none.
    if ctx.headers().get_str(RETRY_COUNT_HEADER).is_none() {
        tracing::info!(payment.id, "processor busy, trying later");
        return HandlerOutcome::retry_after(RETRY_DELAY);
    }
    tracing::info!(payment.id, "charged");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("billing", "0.1.0")).with_broker(
        FileBroker::new("/var/lib/shop/payments.ss"),
        |b| {
            b.include(charge);
        },
    )
}