Operations

A receive operation reports the QoS its subscription reads at; a send operation reports the QoS and retain flag of its policy.

The operation binding carries qos, the level as a number: 0, 1 or 2. A receive operation takes it from the descriptor's qos. A share group is part of the subscribed filter and adds nothing to the binding.

A send operation comes from an Out slot or a dead-letter topic. It reports both arguments of the PUBLISH packet its publish policy declares: qos and retain. A reply has no send operation of its own, so a reply policy adds nothing to the document's operations.

Example

The operation receive_devices___telemetry reports { "qos": 2 }. The slot's operation send_devices___telemetry_alerts_overheat reports { "qos": 1, "retain": true }, from the policy Publish::default().retain(true) the app attaches.

Macros

examples/mqtt_asyncapi.rs

/// Reads the fleet at `QoS` 2 as a member of `workers`, and raises an alert through its slot.
#[subscriber(MqttFilter::new("devices/+/telemetry").qos(Qos::ExactlyOnce).shared("workers"))]
async fn collect(
    telemetry: &Telemetry,
    Out(alerts): Out<impl Publisher, Alerts>,
) -> HandlerOutcome {
    if telemetry.temperature > 30.0 {
        let alert = Alert {
            device: telemetry.device.clone(),
        };
        if let Err(err) = alerts.message(&alert).publish().await {
            tracing::warn!(subscription = "devices/+/telemetry", error = %err, "alert not published");
            return HandlerOutcome::retry();
        }
    }
    HandlerOutcome::ack()
}

Manual

examples/manual/mqtt_asyncapi.rs

/// The fleet, read at `QoS` 2 as a member of `workers`.
fn fleet() -> MqttFilter {
    MqttFilter::new("devices/+/telemetry")
        .qos(Qos::ExactlyOnce)
        .shared("workers")
}