out_retry

A pattern names no subject of its own, so out_retry(Publish).to(subject) says where its delayed copies go; without it the app refuses to start.

A CoreWildcard reads every subject its pattern matches, and a copy has to go to one of them. This crate does not take the pattern itself as a destination, so the mount site names the subject. A registration over a pattern that names no subject fails the start with an error naming the subscription. The service reports the gap at startup instead of losing every delayed message once it runs. out_retry binds the Retry slot, the one the framework publishes its copies through, to a policy; Publish is plain Core NATS. .to(subject) names the subject; a publish transform that declares Names can choose one per delivery instead. A subject the pattern matches brings the copy back to the same handler. The dead-letter subject stays outside the pattern, or the spent message comes back too. A CoreSubject and a JetStreamSubject name their own destination, so they need no out_retry; on them the slot only changes the policy or adds a transform.

Example

Entries of every region are audited again a minute later through audits.retry, which the pattern reads.

Macros
Manual

examples/nats_settlement_out_retry.rs
/// Audits the entries of every region, `audits.eu`, `audits.us` and the rest.
#[subscriber(CoreWildcard::new("audits.*"))]
async fn audit(entry: &Entry) -> HandlerOutcome {
    if entry.posted {
        return HandlerOutcome::ack();
    }
    tracing::info!(entry.id, "not posted yet, auditing again later");
    HandlerOutcome::retry_after(RECHECK)
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("audit", "0.1.0")).with_broker(
        NatsBroker::new("nats://localhost:4222"),
        |b| {
            // The copies go to "audits.retry", which the pattern reads, so they come back here.
            // The dead-letter subject is outside the pattern, so a spent entry does not.
            b.include(audit)
                .max_attempts(nonzero!(3u32))
                .dead_letter("dead.audits")
                .out_retry(Publish)
                .to("audits.retry");
        },
    )
}

examples/manual/nats_settlement_out_retry.rs
/// Audits the entries of every region, `audits.eu`, `audits.us` and the rest.
struct Audit;

impl Handle<Entry> for Audit {
    fn handle(
        &self,
        entry: &Entry,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        if entry.posted {
            return ready(Ok(()));
        }
        tracing::info!(entry.id, "not posted yet, auditing again later");
        ready(Err(HandlerOutcome::retry_after(RECHECK)))
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("audit", "0.1.0")).with_broker(
        NatsBroker::new("nats://localhost:4222"),
        |b| {
            // The copies go to "audits.retry", which the pattern reads, so they come back here.
            // The dead-letter subject is outside the pattern, so a spent entry does not.
            b.include(subscriber(CoreWildcard::new("audits.*"), Audit).build())
                .max_attempts(nonzero!(3u32))
                .dead_letter("dead.audits")
                .out_retry(Publish)
                .to("audits.retry");
        },
    )
}

Example​

Example