out_retry

Each descriptor names where a retry copy goes; a pattern names none, so its registration names the channel with out_retry.

The runtime publishes a copy for a retry_after, for a retry under a cap, and for a dead letter. A stream, a list and a channel subscription each answer with their own key or channel, so the copy reaches the same subscription again. A copy leaves through RedisDefaultPublish unless the registration names another policy. That publisher writes a name the way the service reads it: XADD to a stream, LPUSH to a list in its framing, PUBLISH to a channel in its mode and framing. .out_retry(policy) replaces the policy, and .transform(..) after it changes each copy on its way out. A pattern reads many channels and names none a PUBLISH can reach. Its registration names the channel with .out_retry(policy).to(channel), or mounts a publish transform that declares Destination = Names and names one per delivery. A pattern registration that names neither does not start, and the error names the subscription and both fixes.

Example

The fraud screen reads every payments.* channel; a retry copy goes to payments.retry, which the pattern reads again.

Macros

examples/redis_acknowledgement_out_retry.rs

#[ruststream::app]
fn app() -> impl App<State = AppState> {
    RustStream::new(AppInfo::new("screening", "0.1.0"))
        .on_startup(async move |()| {
            let fraud = FraudCheck {
                endpoint: "https://fraud.internal".into(),
            };
            Ok::<_, Infallible>(AppState { fraud })
        })
        .with_broker(RedisBroker::standalone("redis://localhost:6379"), |b| {
            // The copy is `PUBLISH`ed on `payments.retry`, which the pattern reads again.
            b.include(screen)
                .out_retry(Publish::default())
                .to("payments.retry");
        })
}

Manual

examples/manual/redis_acknowledgement_out_retry.rs

fn app() -> impl App<State = AppState> {
    RustStream::new(AppInfo::new("screening", "0.1.0"))
        .on_startup(async move |()| Ok::<_, Infallible>(startup()))
        .with_broker(RedisBroker::standalone("redis://localhost:6379"), |b| {
            // The copy is `PUBLISH`ed on `payments.retry`, which the pattern reads again.
            let payments = RedisPubSubPattern::new("payments.*");
            b.include(subscriber(payments, Screen).build())
                .out_retry(Publish::default())
                .to("payments.retry");
        })
}