MqttPublishSteps

The qos and retain steps change one message's QoS or retain flag over the policy; a handler bounds its slot with MqttPublishOptions.

MqttPublishSteps puts both steps on the publish builder, in any order. An argument the call does not name keeps what the policy declared at the mount site. The values reach the client as the protocol fields they are, so a subscriber receives a plain message with no extra property. A publish with no call site of its own, a reply or a retry copy, takes the policy whole. The steps are reached from a slot whose bound names the options type: Out<impl Publisher<Options = MqttPublishOptions>, Slot>. That bound and this crate's prelude are the one place a handler body names the broker. A test reads what the steps set with tb.out::<Slot>().with_options(&MqttPublishOptions::default().retain(true)).

Example

A reading changes the state of its device, and the handler publishes that one state retained:

Macros
Manual

examples/mqtt_retained.rs
/// A reading is one of many; the state it puts the device in is the one a service joining later
/// must see. So the handler publishes the state retained, and only the state: the step belongs
/// to this message, not to the publisher.
///
/// Naming a per-message setting is what ties a body to a broker, and its signature says so: the
/// options type is this crate's, and so is the prelude the file imports.
#[subscriber(MqttFilter::new("devices/+/telemetry").qos(Qos::AtLeastOnce))]
async fn announce_state(
    telemetry: &Telemetry,
    Out(states): Out<impl Publisher<Options = MqttPublishOptions>, States>,
) -> HandlerOutcome {
    let state = if telemetry.temperature > 30.0 {
        "hot"
    } else {
        "ok"
    };
    if let Err(err) = states
        .message(&DeviceState(state.as_bytes().to_vec()))
        .retain(true)
        .to()
        .device(&telemetry.device)
        .publish()
        .await
    {
        tracing::warn!(telemetry.device, %err, "state not announced");
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

examples/manual/mqtt_retained.rs
/// A reading is one of many; the state it puts the device in is the one a service joining later
/// must see. So the handler publishes the state retained, and only the state: the step belongs
/// to this message, not to the publisher.
///
/// Naming a per-message setting is what ties a body to a broker, and its bound says so: the
/// options type is this crate's, and so is the prelude the file imports.
struct AnnounceState;

impl<Entry> Handle<Telemetry, (), Outs<(Entry,)>> for AnnounceState
where
    Entry: OutEntry<States, Wire: Publisher<Options = MqttPublishOptions>>,
{
    async fn handle(
        &self,
        telemetry: &Telemetry,
        outs: &Outs<(Entry,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let state = if telemetry.temperature > 30.0 {
            "hot"
        } else {
            "ok"
        };
        if let Err(err) = outs
            .get(States)
            .message(&DeviceState(state.as_bytes().to_vec()))
            .retain(true)
            .to()
            .device(&telemetry.device)
            .publish()
            .await
        {
            tracing::warn!(telemetry.device, %err, "state not announced");
            return Err(HandlerOutcome::retry());
        }
        Ok(())
    }
}

fn fleet() -> MqttFilter {
    MqttFilter::new("devices/+/telemetry").qos(Qos::AtLeastOnce)
}

The mount site fixes the slot's policy, which the step changes for one message:

Macros
Manual

examples/mqtt_retained.rs
#[ruststream::app]
fn app() -> impl App {
    service(MqttBroker::new("mqtt://localhost:1883", "state-tracker"))
}

fn service(broker: MqttBroker) -> RustStream {
    RustStream::new(AppInfo::new("state-tracker", "0.1.0")).with_broker(broker, |b| {
        // Every publish of this policy is retained, so a dashboard that starts later still
        // reads the status.
        b.after_startup(
            Publish::default().qos(Qos::AtLeastOnce).retain(true),
            async move |publisher| -> Result<(), PublishError<MqttError>> {
                publisher
                    .message(&ServiceStatus(b"online".to_vec()))
                    .publish()
                    .await
            },
        );
        // The mount site fixes what every publish through the slot takes; the handler's step is
        // what one message changes.
        b.include(announce_state)
            .out(States, Publish::default().qos(Qos::AtLeastOnce))
            .out_retry(Publish::default())
            .to("devices/retry/telemetry")
            .build();
    })
}

examples/manual/mqtt_retained.rs
fn app() -> impl App {
    service(MqttBroker::new("mqtt://localhost:1883", "state-tracker"))
}

fn service(broker: MqttBroker) -> RustStream {
    RustStream::new(AppInfo::new("state-tracker", "0.1.0")).with_broker(broker, |b| {
        // Every publish of this policy is retained, so a dashboard that starts later still
        // reads the status.
        b.after_startup(
            Publish::default().qos(Qos::AtLeastOnce).retain(true),
            async move |publisher| -> Result<(), PublishError<MqttError>> {
                publisher
                    .message(&ServiceStatus(b"online".to_vec()))
                    .publish()
                    .await
            },
        );
        // The mount site fixes what every publish through the slot takes; the handler's step is
        // what one message changes.
        b.include(
            subscriber("devices/+/telemetry", AnnounceState)
                .map_source(|_| fleet())
                .build(),
        )
        .out(States, Publish::default().qos(Qos::AtLeastOnce))
        .out_retry(Publish::default())
        .to("devices/retry/telemetry")
        .build();
    })
}

Example​

Example