MqttContext

Ctx<DeliveryTopic> gives a handler the topic a message was published to, which differs from the filter that matched it.

MqttContext is what a delivery carries besides its payload and headers. The subscription builds it once per delivery. It holds one field, the topic, under the key DeliveryTopic. Both are in the prelude. A handler binds the topic as a parameter with Ctx<DeliveryTopic>, which hands it over as a String. A body that takes ctx: &mut Context<'_, MqttContext> reads it borrowed with ctx.context(DeliveryTopic), as the manual form does.

On an MqttFilter the topic tells the messages of the filter apart: devices/dev42/telemetry and devices/dev43/telemetry both match devices/+/telemetry. The subscription's name, ctx.name(), is the filter, never the topic. The same key reads the topic in a publish transform on the reply or the retry position, through cx.context(DeliveryTopic); a transform on the retry position sends a copy back to the topic its delivery arrived on.

Example

An alarm names no device; the topic it arrived on does.

Macros
Manual

examples/mqtt_context.rs
/// The alarm itself names no device: its topic does, at the level the filter's `+` matched.
#[subscriber(MqttFilter::new("devices/+/alarms"), reply)]
async fn open_incident(alarm: &Alarm, Ctx(topic): Ctx<DeliveryTopic>) -> Incident {
    let device = topic.split('/').nth(1).unwrap_or_default().to_owned();
    tracing::info!(%topic, alarm.code, "incident opened");
    Incident {
        device,
        code: alarm.code,
    }
}

examples/manual/mqtt_context.rs
/// The alarm itself names no device: its topic does, at the level the filter's `+` matched.
struct OpenIncident;

impl Handle<Alarm, Incident, (), MqttContext> for OpenIncident {
    fn handle(
        &self,
        alarm: &Alarm,
        _outs: &(),
        ctx: &mut Context<'_, MqttContext>,
    ) -> impl Future<Output = Result<Incident, HandlerOutcome>> {
        let topic = ctx.context(DeliveryTopic);
        let device = topic.split('/').nth(1).unwrap_or_default().to_owned();
        tracing::info!(%topic, alarm.code, "incident opened");
        ready(Ok(Incident {
            device,
            code: alarm.code,
        }))
    }
}

Example​

Example