Retries

max_attempts(n) caps the deliveries one message gets, counted on the retry-count header, because MQTT counts no redeliveries.

A redelivered PUBLISH packet carries a duplicate flag and no counter. So the runtime counts on RETRY_COUNT_HEADER, which travels on the copies retry_after publishes. The cap counts deliveries, the first one included: nonzero!(5u32) gives a message five. Under a cap every retry travels as such a copy, an immediate retry() too, so the count reaches the cap whichever outcome the handler returns. The delivery that reaches the cap goes to the dead-letter topic instead of coming back. A cap without a dead-letter topic rejects the delivery, which on MQTT acknowledges it and lets it go; a warning names the subscription. A filter subscription also names where its copies go, with out_retry.

Example

Every registration of the service gets five deliveries per message and a dead-letter topic of its own.

Macros
Manual

examples/mqtt_retries.rs
#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("telemetry-retries", "0.1.0")).with_broker(
        MqttBroker::new("mqtt://localhost:1883", "telemetry-retries"),
        |b| {
            // Five deliveries of one reading, the first included; the fifth failure goes to
            // "dead/readings" instead, payload and headers as they arrived.
            b.include(store)
                .max_attempts(nonzero!(5u32))
                .dead_letter("dead/readings");
            b.include(on_dead_letter);
            // A filter names where its copies go: here one topic the filter matches, so they
            // come back to this subscription.
            b.include(collect)
                .max_attempts(nonzero!(5u32))
                .dead_letter("dead/telemetry")
                .out_retry(Publish::default())
                .to("devices/retry/telemetry");
            // Or per delivery: each copy returns to the topic it arrived on.
            b.include(track)
                .max_attempts(nonzero!(5u32))
                .dead_letter("dead/sites")
                .out_retry(Publish::default())
                .transform(ToDeliveryTopic);
        },
    )
}

examples/manual/mqtt_retries.rs
fn app() -> impl App {
    RustStream::new(AppInfo::new("telemetry-retries", "0.1.0")).with_broker(
        MqttBroker::new("mqtt://localhost:1883", "telemetry-retries"),
        |b| {
            // Five deliveries of one reading, the first included; the fifth failure goes to
            // "dead/readings" instead, payload and headers as they arrived.
            b.include(
                subscriber("meters/readings", Store)
                    .map_source(|_| readings())
                    .build(),
            )
            .max_attempts(nonzero!(5u32))
            .dead_letter("dead/readings");
            b.include(
                subscriber("dead/readings", OnDeadLetter)
                    .map_source(|_| dead_readings())
                    .build(),
            );
            // A filter names where its copies go: here one topic the filter matches, so they
            // come back to this subscription.
            b.include(
                subscriber("devices/+/telemetry", Collect)
                    .map_source(|_| fleet())
                    .build(),
            )
            .max_attempts(nonzero!(5u32))
            .dead_letter("dead/telemetry")
            .out_retry(Publish::default())
            .to("devices/retry/telemetry");
            // Or per delivery: each copy returns to the topic it arrived on.
            b.include(
                subscriber("sites/+/telemetry", Track)
                    .map_source(|_| sites())
                    .build(),
            )
            .max_attempts(nonzero!(5u32))
            .dead_letter("dead/sites")
            .out_retry(Publish::default())
            .transform(ToDeliveryTopic);
        },
    )
}

Example​

Example