Перейти к основному содержимому
Эта страница ещё не переведена

Ниже показан английский оригинал.

dead_letter

max_attempts(n) caps deliveries and dead_letter(name) names where a spent message goes, written with its own form's command.

Both are steps of the mount chain, and they read the same on a stream, a list and a channel. max_attempts(n) counts every delivery of one message, the first one included. When the n-th delivery asks for a retry, the runtime settles it and publishes it to the dead-letter name instead, payload and headers as they arrived. The name is written the way the source form writes: XADD from a stream, LPUSH from a list, PUBLISH from a channel. A cap without a dead-letter name drops the spent delivery and logs a warning that names the subscription. A dead-letter name without a cap carries every retry away on the first one. Under a cap an immediate retry travels as a copy the runtime publishes, so that the count moves with the message in RETRY_COUNT_HEADER. A claiming subscription reads Redis's own delivery count instead and appends nothing. Redis keeps a delivery count only in a stream's pending entries list, which the claiming and reclaim modes read. On a list, a channel and a plain stream read, the runtime applies the cap with the count its copies carry. The dead-letter key is an ordinary key of the same broker, and a subscription reads it like any other. A name one subscription reads as one Redis type and another writes as a second type stops the service at startup, and the error names both.

Example​

A payment the gateway keeps rejecting goes to payments.dlq after five deliveries; a refund goes to refunds.failed after three.

examples/redis_acknowledgement_dead_letter.rs
#[ruststream::app]
fn app() -> impl App<State = AppState> {
    RustStream::new(AppInfo::new("billing", "0.1.0"))
        .on_startup(async move |()| {
            let gateway = PaymentGateway {
                endpoint: "https://gateway.internal".into(),
            };
            Ok::<_, Infallible>(AppState { gateway })
        })
        .with_broker(RedisBroker::standalone("redis://localhost:6379"), |b| {
            // Five deliveries per payment, the first included. The fifth failure goes to the
            // `payments.dlq` stream with `XADD`, payload and headers as they arrived.
            b.include(charge)
                .max_attempts(nonzero!(5u32))
                .dead_letter("payments.dlq");
            // Three deliveries per refund; the third failure is `LPUSH`ed onto `refunds.failed`.
            b.include(refund)
                .max_attempts(nonzero!(3u32))
                .dead_letter("refunds.failed");
            b.include(on_dead_letter);
        })
}

An operator's subscription reads the dead-letter stream:

examples/redis_acknowledgement_dead_letter.rs
/// The dead-letter stream is read by an ordinary subscription. The retry-count header says how
/// many times the payment was delivered before it was carried here.
#[subscriber(RedisStream::new("payments.dlq").group("operators"))]
async fn on_dead_letter(payment: &Payment, ctx: &mut Context<'_>) -> HandlerOutcome {
    let deliveries = ctx.headers().get_str(RETRY_COUNT_HEADER).unwrap_or("1");
    tracing::error!(payment.id, deliveries, "payment dead-lettered");
    HandlerOutcome::ack()
}