Dead letters

dead_letter(topic) names the topic a delivery that reached its cap is published to; it lies outside every filter the service reads.

It follows max_attempts(n) at the mount site. When the n-th delivery asks for a retry, the runtime acknowledges it and publishes it to the dead-letter topic instead. The message keeps its payload and its headers. RETRY_COUNT_HEADER on it says how many times it was delivered. The dead-letter topic is an ordinary topic, and a subscription of this service or of another one reads it. A filter of the service that matches the dead-letter topic receives the spent message as a new delivery, so the topic stays outside every other filter the service reads.

Example

Each registration sends its spent messages to a topic of its own under dead/:

Macros

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);
        },
    )
}

Manual

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);
        },
    )
}

The dead-letter topic is read by an ordinary subscription:

Macros

examples/mqtt_retries.rs

/// The dead-letter topic is read by an ordinary subscription. The retry-count header says how
/// many times the reading was delivered before it was carried here.
#[subscriber(MqttTopic::new("dead/readings").qos(Qos::AtLeastOnce))]
async fn on_dead_letter(reading: &Reading, ctx: &mut Context) -> HandlerOutcome {
    let deliveries = ctx.headers().get_str(RETRY_COUNT_HEADER).unwrap_or("1");
    tracing::error!(reading.meter, deliveries, "reading dead-lettered");
    HandlerOutcome::ack()
}

Manual

examples/manual/mqtt_retries.rs

/// The dead-letter topic is read by an ordinary subscription. The retry-count header says how
/// many times the reading was delivered before it was carried here.
struct OnDeadLetter;

impl Handle<Reading> for OnDeadLetter {
    fn handle(
        &self,
        reading: &Reading,
        _outs: &(),
        ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        let deliveries = ctx.headers().get_str(RETRY_COUNT_HEADER).unwrap_or("1");
        tracing::error!(reading.meter, deliveries, "reading dead-lettered");
        ready(Ok(()))
    }
}

fn dead_readings() -> MqttTopic {
    MqttTopic::new("dead/readings").qos(Qos::AtLeastOnce)
}