Skip to main content

out_retry

A filter subscription names where its retry copies go, once with out_retry(..).to(..) or per delivery with a transform.

An MqttTopic reads one topic, and its copies go there; a share group's copies go to the plain topic, not to $share/... An MqttFilter matches many topics and is a topic of its own only by accident, so the mount site names the destination. out_retry binds the Retry slot, the one the runtime publishes its copies through, to a publish policy. .to(topic) sends every copy to one topic; a topic the filter matches brings the copies back to the same handler. A publish transform that declares Names picks the topic per delivery instead, such as the topic the delivery arrived on. It reads that topic from the MqttContext of the delivery, which reaches the slot when the handler reads it: a Ctx<DeliveryTopic> parameter, or a manual body typed over MqttContext. Naming a topic and adding a naming transform are mutually exclusive, and writing both does not compile. A filter registration that names neither does not start, and the error names the subscription. The copy carries the delivery's own bytes, so a codec on the slot encodes nothing, while a transform still runs.

Example​

A shared fleet subscription sends its copies to one retry topic its filter matches:

examples/mqtt_retries.rs
/// A filter over the whole fleet, shared between the instances of the service. A filter is no
/// name a publisher can use, so the mount site names the topic its copies go to.
#[subscriber(MqttFilter::new("devices/+/telemetry").qos(Qos::AtLeastOnce).shared("workers"))]
async fn collect(reading: &Reading) -> HandlerOutcome {
    if let Err(err) = write(reading).await {
        tracing::warn!(reading.meter, %err, "write failed, retrying");
        return HandlerOutcome::retry_after(BACKOFF);
    }
    HandlerOutcome::ack()
}

A site subscription sends each copy back to the topic it arrived on:

examples/mqtt_retries.rs
/// Sends every copy back to the topic its delivery arrived on, the one topic of the filter's many
/// that this message belongs to. A transform on the retry position reads the delivery being
/// retried, and the topic lives in the broker's per-delivery context.
struct ToDeliveryTopic;

impl<Options> PublishTransform<ForReply<MqttContext>, Options> for ToDeliveryTopic {
    type Destination = Names;

    fn apply(
        &self,
        out: &mut Outgoing<'_>,
        _options: &mut Option<Options>,
        cx: &PublishContext<'_, MqttContext>,
    ) {
        out.set_name(cx.context(DeliveryTopic).to_owned());
    }
}

/// The `Ctx<DeliveryTopic>` parameter hands the body the topic and makes the crate's context the
/// registration's, which is what the transform reads.
#[subscriber(MqttFilter::new("sites/+/telemetry").qos(Qos::AtLeastOnce))]
async fn track(reading: &Reading, Ctx(topic): Ctx<DeliveryTopic>) -> HandlerOutcome {
    if let Err(err) = write(reading).await {
        tracing::warn!(reading.meter, %topic, %err, "write failed, retrying");
        return HandlerOutcome::retry_after(BACKOFF);
    }
    HandlerOutcome::ack()
}

The mount site names the destination of each filter subscription:

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