Acknowledgement

An ack sends PUBACK at QoS 1 and PUBREC at QoS 2; QoS 0 has no acknowledgement, and MQTT has no negative one.

A message settles when the handler returns, and the delivery's own quality of service decides what that sends. A delivery arrives at the lower of the publisher's QoS and the subscription's, so a QoS 1 subscription still receives a QoS 0 publish at QoS 0. At QoS 1 and 2 the server keeps a delivery until it is acknowledged; the client finishes the QoS 2 handshake on its own. A QoS 0 delivery has no acknowledgement packet, so settling it reports AckError::Unsupported, which the runtime logs as a warning. drop acknowledges the delivery too, because an acknowledgement is the only final answer MQTT has; the server never sends the message again. retry asks for a negative acknowledgement, which MQTT does not have, so the delivery stays unacknowledged. retry_after is the outcome that retries within the session. In the manual form Ok acknowledges and Err(outcome) settles by the outcome. When the server sends one packet for two overlapping filters, the acknowledgement belongs to one delivery, and settling the other reports Unsupported.

Example

A device agent applies the commands it knows and drops the rest; both answers settle the delivery.

Macros
Manual

examples/mqtt_settlement_ack.rs
/// The subscription reads at `QoS` 1, so returning from the handler acknowledges the delivery with
/// a PUBACK. A command the device does not know is dropped: on MQTT that acknowledges it as well,
/// and the server never sends it again.
#[subscriber(MqttTopic::new("devices/dev42/commands").qos(Qos::AtLeastOnce))]
async fn apply(command: &Command) -> HandlerOutcome {
    match command.action.as_str() {
        "reboot" | "reset" => {
            tracing::info!(command.action, "command applied");
            HandlerOutcome::ack()
        }
        unknown => {
            tracing::warn!(action = unknown, "unknown command dropped");
            HandlerOutcome::drop()
        }
    }
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("device-agent", "0.1.0")).with_broker(
        MqttBroker::new("mqtt://localhost:1883", "device-agent-dev42"),
        |b| {
            b.include(apply);
        },
    )
}

examples/manual/mqtt_settlement_ack.rs
/// The subscription reads at `QoS` 1, so returning from the handler acknowledges the delivery with
/// a PUBACK. A command the device does not know is dropped: on MQTT that acknowledges it as well,
/// and the server never sends it again.
struct Apply;

impl Handle<Command> for Apply {
    fn handle(
        &self,
        command: &Command,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        let verdict = match command.action.as_str() {
            "reboot" | "reset" => {
                tracing::info!(command.action, "command applied");
                Ok(())
            }
            unknown => {
                tracing::warn!(action = unknown, "unknown command dropped");
                Err(HandlerOutcome::drop())
            }
        };
        ready(verdict)
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("device-agent", "0.1.0")).with_broker(
        MqttBroker::new("mqtt://localhost:1883", "device-agent-dev42"),
        |b| {
            let commands = MqttTopic::new("devices/dev42/commands").qos(Qos::AtLeastOnce);
            b.include(
                subscriber("devices/dev42/commands", Apply)
                    .map_source(|_| commands)
                    .build(),
            );
        },
    )
}

Example​

Example