Перейти к основному содержимому
Нормативная версия документации - английская

Эта страница переведена с английского языковой моделью. При любом расхождении верен английский оригинал.

Состояние устройств: сохранённые сообщения и последняя воля

Это руководство хранит текущее состояние каждого устройства на сервере, чтобы читатель, который подключается позже, сразу его видел, и сообщает, работает ли сам сервис.

Истории MQTT не хранит, за одним исключением: сохранённого сообщения. Сервер хранит последнее сохранённое сообщение каждой темы и отправляет его каждой новой подписке, фильтр которой подходит к этой теме. Последняя воля - это сообщение, которое сервер публикует за клиента, если его соединение завершилось без штатного отключения.

Пример - монитор шлюзов. Шлюзы периодически присылают сигналы присутствия (heartbeat) с уровнем заряда батареи. Сервис хранит состояние каждого шлюза, ok или low-battery, сохранённым сообщением в gateways/<id>/state, а панели мониторинга читают его оттуда.

Состояние каждого шлюза​

src/main.rs
/// The state of one gateway, as plain text a dashboard shows as it is. The type carries its own
/// bytes, so no codec runs on them, and the `{gateway}` level becomes a setter.
#[derive(Outgoing, Serialized)]
#[outgoing(name = "gateways/{gateway}/state")]
struct GatewayState(Vec<u8>);

#[derive(OutSlot)]
#[publishes(GatewayState)]
struct States;

/// Turns each heartbeat into the gateway's state.
#[subscriber(MqttFilter::new("gateways/+/heartbeat").qos(Qos::AtLeastOnce))]
async fn track(heartbeat: &Heartbeat, Out(states): Out<impl Publisher, States>) -> HandlerOutcome {
    let state = if heartbeat.battery_percent < 20 {
        "low-battery"
    } else {
        "ok"
    };
    let announced = states
        .message(&GatewayState(state.as_bytes().to_vec()))
        .to()
        .gateway(&heartbeat.gateway)
        .publish()
        .await;
    if let Err(err) = announced {
        tracing::warn!(heartbeat.gateway, %err, "state not announced");
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

/// An empty retained payload deletes the retained message, so a removed gateway leaves no state.
#[subscriber("fleet/decommissioned")]
async fn forget(gone: &Decommissioned, Out(states): Out<impl Publisher, States>) -> HandlerOutcome {
    let cleared = states
        .message(&GatewayState(Vec::new()))
        .to()
        .gateway(&gone.gateway)
        .publish()
        .await;
    if let Err(err) = cleared {
        tracing::warn!(gone.gateway, %err, "state not cleared");
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

Состояние - это простой текст, который панель показывает как есть. Поэтому у GatewayState стоит #[derive(Serialized)], тип сам хранит свои байты, а кодек к ним не применяется. Его тема - шаблон имени: {gateway} становится сеттером, поэтому .to().gateway(..) указывает тему одного шлюза. track публикует состояние шлюза, от которого пришёл сигнал.

forget публикует пустую полезную нагрузку в тему состояния шлюза, который выводят из эксплуатации. Пустое сохранённое сообщение удаляет сохранённое сообщение своей темы, и сервер перестаёт отдавать это состояние новым подписчикам.

Сохранённые публикации​

src/main.rs
fn service(broker: MqttBroker) -> RustStream {
    RustStream::new(AppInfo::new("gateway-monitor", "0.1.0")).with_broker(broker, |b| {
        // Runs once the broker is connected, and replaces the will's `offline` of an earlier run.
        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
            },
        );
        // Every state goes out retained, so a dashboard that starts later reads it at once.
        let retained = Publish::default().qos(Qos::AtLeastOnce).retain(true);
        b.include(track)
            .out(States, retained)
            .out_retry(Publish::default())
            .to("gateways/retry/heartbeat")
            .build();
        b.include(forget).out(States, retained).build();
    })
}

#[ruststream::app]
fn app() -> impl App<State = ()> {
    service(broker("mqtt://localhost:1883", "gateway-monitor"))
}

retain(true) в политике публикации делает сохранённой каждую публикацию через слот. Оба обработчика используют слот States и одну политику, поэтому состояние и его удаление публикуются одинаково. Сохранённым можно сделать и отдельное сообщение: шаг retain у публикации переопределяет политику для этого вызова.

Совместная подписка не получает сохранённых сообщений, когда подписывается. Панель, которая читает состояния, подписывается вне любой группы совместной подписки.

В сети и не в сети​

src/main.rs
/// The service's own status, which dashboards read to tell whether it is running.
#[derive(Outgoing, Serialized)]
#[outgoing(name = "services/gateway-monitor/status")]
struct ServiceStatus(Vec<u8>);

/// If the connection ends without a clean disconnect, the server publishes `offline` on the
/// status topic, retained.
fn broker(url: &str, client_id: &str) -> MqttBroker {
    MqttBroker::new(url, client_id).last_will(
        "services/gateway-monitor/status",
        "offline",
        Qos::AtLeastOnce,
        true,
    )
}

last_will принимает тему, полезную нагрузку, качество обслуживания и флаг сохранения последней воли. Брокер передаёт последнюю волю серверу при подключении. Если соединение завершается без пакета протокола DISCONNECT, потому что процесс аварийно завершился или отказала сеть, сервер публикует последнюю волю: здесь offline, сохранённым сообщением в services/gateway-monitor/status. Штатная остановка отправляет DISCONNECT, и сервер отбрасывает последнюю волю.

Вторая половина - хук after_startup в приложении выше. Он выполняется, когда брокер подключён, и публикует online сохранённым сообщением в ту же тему. Поэтому в теме статуса всегда лежит последнее из двух, и подписавшаяся панель сразу его читает.

Проверка состояния​

Внутрипроцессные тесты проверяют, что публикует сервис: состояние каждого шлюза, пустую полезную нагрузку для удалённого шлюза и online при запуске. Внутрипроцессный сервер хранит сохранённые сообщения так же, как Mosquitto. Последнюю волю публикует сервер, когда соединение обрывается, поэтому один тест работает на Mosquitto:

MQTT_TEST_URL=mqtt://localhost:1883 cargo test -- --test-threads=1

Он запускает сервис, штатно останавливает его и после этого запускает панель: панель получает online и состояние шлюза, оба сохранённые. Затем он завершает очередной запуск сервиса без DISCONNECT, и панель получает offline. В конце он удаляет шлюз теста, и его состояние на сервере стирается.