retry

retry() leaves a delivery unacknowledged: it comes back when a persistent session resumes, never within the connection.

The outcome asks the broker for a negative acknowledgement, and MQTT has none. The runtime logs the refusal and moves on, and nothing acknowledges the delivery later. At QoS 1 and 2 the server keeps the message in the client's session. It sends the message again when the service connects with the same client id and the session is still there: clean_start(false) and a session_expiry longer than the gap. With a clean start, or once the session expired, the message is gone; at QoS 0 it is gone at once. Until the connection ends, the delivery holds one slot of the receive_maximum window, 1000 by default. When the window is full of unacknowledged deliveries, the server sends no more QoS 1 or 2 messages on that connection. For a retry within the session, a handler returns retry_after. A registration with a cap turns retry() into an immediate copy of the message instead.

Example

An update that needs a reboot waits for the next session of the device agent.

Macros

examples/mqtt_settlement_retry.rs

/// An update that needs a reboot waits for the next session: the handler leaves it
/// unacknowledged, and the server hands it back when the device agent reconnects.
#[subscriber(MqttTopic::new("devices/dev42/firmware").qos(Qos::AtLeastOnce))]
async fn install(update: &Firmware) -> HandlerOutcome {
    if update.needs_reboot {
        tracing::info!(update.version, "update left for the next session");
        return HandlerOutcome::retry();
    }
    tracing::info!(update.version, "update installed");
    HandlerOutcome::ack()
}

Manual

examples/manual/mqtt_settlement_retry.rs

/// An update that needs a reboot waits for the next session: the handler leaves it
/// unacknowledged, and the server hands it back when the device agent reconnects.
struct Install;

impl Handle<Firmware> for Install {
    fn handle(
        &self,
        update: &Firmware,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        if update.needs_reboot {
            tracing::info!(update.version, "update left for the next session");
            return ready(Err(HandlerOutcome::retry()));
        }
        tracing::info!(update.version, "update installed");
        ready(Ok(()))
    }
}

fn firmware() -> MqttTopic {
    MqttTopic::new("devices/dev42/firmware").qos(Qos::AtLeastOnce)
}

The broker keeps the session for a day after the connection ends:

Macros

examples/mqtt_settlement_retry.rs

#[ruststream::app]
fn app() -> impl App {
    service(MqttBroker::new(
        "mqtt://localhost:1883",
        "device-agent-dev42",
    ))
}

/// The session outlives the connection by a day, so what the handler left unacknowledged comes
/// back when the agent connects again within it, under the same client id.
fn service(broker: MqttBroker) -> RustStream {
    RustStream::new(AppInfo::new("device-agent", "0.1.0")).with_broker(
        broker
            .clean_start(false)
            .session_expiry(Duration::from_hours(24)),
        |b| {
            b.include(install);
        },
    )
}

Manual

examples/manual/mqtt_settlement_retry.rs

fn app() -> impl App {
    service(MqttBroker::new(
        "mqtt://localhost:1883",
        "device-agent-dev42",
    ))
}

/// The session outlives the connection by a day, so what the handler left unacknowledged comes
/// back when the agent connects again within it, under the same client id.
fn service(broker: MqttBroker) -> RustStream {
    RustStream::new(AppInfo::new("device-agent", "0.1.0")).with_broker(
        broker
            .clean_start(false)
            .session_expiry(Duration::from_hours(24)),
        |b| {
            b.include(
                subscriber("devices/dev42/firmware", Install)
                    .map_source(|_| firmware())
                    .build(),
            );
        },
    )
}