out_retry

On a pipe a deferred copy goes to the next stage, so the mount site names it with out_retry(Publish).to(key) or the service does not start.

out_retry(policy) names the publisher the runtime's deferred copies leave through, and .to(key) names the stream key they go to. Standard output reaches the next process in the pipeline and never this one's standard input. So a copy cannot come back to the stage that deferred it, and the stage says where it goes. Every registration on StdioBroker names it, because a handler may answer retry_after on any delivery. A registration that names none stops the service at startup, with an error that names the subscription and the .to(..) step that fixes it. The copy leaves once the delay is over, and a downstream stage reads it under that key, for example to feed it back to the start of the pipeline. On a stream file the copy needs no destination (retry_after), and out_retry(Publish) there only replaces the publisher. The position is a slot, so .transform(..) runs on the copy; the copy carries the delivery's own bytes, so no codec encodes it.

Example

The stage hands every job it cannot run now to jobs.retry after five seconds.

Macros
Manual

examples/stdio_out_retry.rs
/// Hands every job it cannot run now to the retry stage, after a pause.
#[subscriber("jobs")]
async fn work(job: &Job) -> HandlerOutcome {
    tracing::info!(job.id, "busy, deferring");
    HandlerOutcome::retry_after(RETRY_DELAY)
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("worker", "0.1.0")).with_broker(StdioBroker::new(), |b| {
        b.include(work).out_retry(Publish).to("jobs.retry");
    })
}

examples/manual/stdio_out_retry.rs
/// Hands every job it cannot run now to the retry stage, after a pause.
struct Work;

impl Handle<Job> for Work {
    fn handle(
        &self,
        job: &Job,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(job.id, "busy, deferring");
        ready(Err(HandlerOutcome::retry_after(RETRY_DELAY)))
    }
}

fn app() -> impl App<State = ()> {
    RustStream::new(AppInfo::new("worker", "0.1.0")).with_broker(StdioBroker::new(), |b| {
        b.include(subscriber("jobs", Work).build())
            .out_retry(Publish)
            .to("jobs.retry");
    })
}

Example​

Example