Темы, подстановочные знаки и совместные подписки

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

Тема - это имя из уровней, разделённых /, например devices/dev42/telemetry. Сервер передаёт сообщение каждой подписке, фильтр которой подходит к его теме в этот момент. Впрок он ничего не хранит, кроме одного сохранённого сообщения на тему (Состояние устройств) и очереди постоянной сессии (Гарантии доставки).

Пример - сервис парка устройств. Он сохраняет телеметрию каждого устройства, трассирует все сообщения одного устройства и планирует раскатку прошивок.

Группа совместной подписки

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

Макросы

src/main.rs

/// Every replica of the service joins the share group `ingest`, so each reading is stored once.
#[subscriber(MqttFilter::new("devices/+/telemetry").qos(Qos::AtLeastOnce).shared("ingest"))]
async fn store(telemetry: &Telemetry, Ctx(topic): Ctx<DeliveryTopic>) -> HandlerOutcome {
    tracing::info!(%topic, telemetry.temperature, "reading stored");
    HandlerOutcome::ack()
}

Вручную

src/main.rs

/// Every replica of the service joins the share group `ingest`, so each reading is stored once.
/// A body typed over `MqttContext` reads the topic the reading arrived on.
struct Store;

impl Handle<Telemetry, (), (), MqttContext> for Store {
    fn handle(
        &self,
        telemetry: &Telemetry,
        _outs: &(),
        ctx: &mut Context<'_, MqttContext>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        let topic = ctx.context(DeliveryTopic);
        tracing::info!(%topic, telemetry.temperature, "reading stored");
        ready(Ok(()))
    }
}

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

MqttFilter читает каждую тему, к которой подходит фильтр. + соответствует ровно одному уровню, поэтому devices/+/telemetry читает телеметрию каждого устройства и ничего глубже. shared("ingest") подписывается на $share/ingest/devices/+/telemetry, и сервер передаёт каждое сообщение одной сессии группы. Каждая реплика подключается со своим идентификатором клиента и входит в ту же группу, поэтому реплики делят показания между собой. Подписка вне группы или в другой группе по-прежнему получает свою копию.

Обработчик не видит ни префикса $share/.., ни самого фильтра. Ctx<DeliveryTopic> - это тема, на которую пришло показание, например devices/dev42/telemetry, поэтому идентификатор устройства доступен обработчику.

Порядок по устройству

Сервер отдаёт каждое показание группы совместной подписки любому её участнику, поэтому показания одного устройства могут попасть на разные реплики. Порядок по устройству сохраняется внутри одной подписки: workers(n, by_key) выполняет показания с общим ключом одно за другим, а с разными ключами - параллельно. Ключ - заголовок partition-key, PARTITION_KEY_HEADER. Устройство ставит его в каждое показание, например свой идентификатор. Подписка, которой нужен порядок показаний каждого устройства, читает вне группы совместной подписки.

Многоуровневый подстановочный знак

trace пишет в журнал каждое сообщение одного устройства, какого бы рода оно ни было.

Макросы

src/main.rs

/// A payload taken as the bytes it is, because one device sends messages of several kinds.
#[derive(Deserialized)]
struct Raw<'a>(&'a [u8]);

/// Logs every message of the device under investigation, whatever its kind.
#[subscriber(MqttFilter::new("devices/dev42/#"))]
async fn trace(message: &Raw<'_>, Ctx(topic): Ctx<DeliveryTopic>) -> HandlerOutcome {
    tracing::debug!(%topic, bytes = message.0.len(), "dev42 traced");
    HandlerOutcome::ack()
}

Вручную

src/main.rs

/// A payload taken as the bytes it is, because one device sends messages of several kinds.
#[derive(Deserialized)]
struct Raw<'a>(&'a [u8]);

/// Logs every message of the device under investigation, whatever its kind.
struct Trace;

impl Handle<Raw<'_>, (), (), MqttContext> for Trace {
    fn handle(
        &self,
        message: &Raw<'_>,
        _outs: &(),
        ctx: &mut Context<'_, MqttContext>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        let topic = ctx.context(DeliveryTopic);
        tracing::debug!(%topic, bytes = message.0.len(), "dev42 traced");
        ready(Ok(()))
    }
}

fn one_device() -> MqttFilter {
    MqttFilter::new("devices/dev42/#")
}

# соответствует остатку темы из любого числа уровней и стоит в фильтре последним. devices/dev42/# читает devices/dev42/telemetry, devices/dev42/telemetry/raw и любую другую тему устройства. Устройство отправляет сообщения нескольких родов, поэтому trace принимает полезную нагрузку как есть, в байтах, и кодек к ней не применяется.

Тема, которая начинается с $, например собственная статистика сервера $SYS/.., подходит только к фильтру, который тоже начинается с $. Поэтому один # её никогда не читает.

Одна тема

Макросы

src/main.rs

#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "fleet/firmware")]
struct FirmwareRelease {
    version: String,
}

/// A bare string is one topic, read at the default quality of service.
#[subscriber("fleet/firmware")]
async fn schedule_rollout(release: &FirmwareRelease) -> HandlerOutcome {
    tracing::info!(release.version, "rollout scheduled");
    HandlerOutcome::ack()
}

Вручную

src/main.rs

#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
#[outgoing(name = "fleet/firmware")]
struct FirmwareRelease {
    version: String,
}

/// Mounted under a bare name, which is one topic read at the default quality of service.
struct ScheduleRollout;

impl Handle<FirmwareRelease> for ScheduleRollout {
    fn handle(
        &self,
        release: &FirmwareRelease,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(release.version, "rollout scheduled");
        ready(Ok(()))
    }
}

Простая строка в #[subscriber(..)] - это одна тема, которая читается с настройками MqttTopic по умолчанию. MqttTopic::new("fleet/firmware").qos(..) записывает ту же подписку с её настройками. MqttTopic не принимает подстановочных знаков: если в имени есть такой знак, запуск возвращает ошибку с упоминанием MqttFilter ещё до любого ввода-вывода. Подстановочные знаки бывают только у подписок, поэтому публикация называет одну тему, а публикация в тему с + или # возвращает ошибку.

Приложение

Макросы

src/main.rs

#[ruststream::app]
fn app() -> impl App<State = ()> {
    RustStream::new(AppInfo::new("fleet", "0.1.0")).with_broker(
        MqttBroker::new("mqtt://localhost:1883", "fleet-svc"),
        |b| {
            // A filter names no topic of its own, so each one names where a retry copy goes:
            // a topic it matches and no other filter of the service does.
            b.include(store)
                .out_retry(Publish::default())
                .to("devices/retry/telemetry");
            b.include(trace)
                .out_retry(Publish::default())
                .to("devices/dev42/retry");
            b.include(schedule_rollout);
        },
    )
}

Вручную

src/main.rs

fn app() -> impl App<State = ()> {
    RustStream::new(AppInfo::new("fleet", "0.1.0")).with_broker(
        MqttBroker::new("mqtt://localhost:1883", "fleet-svc"),
        |b| {
            // A filter names no topic of its own, so each one names where a retry copy goes:
            // a topic it matches and no other filter of the service does.
            b.include(
                subscriber("devices/+/telemetry", Store)
                    .map_source(|_| fleet())
                    .build(),
            )
            .out_retry(Publish::default())
            .to("devices/retry/telemetry");
            b.include(
                subscriber("devices/dev42/#", Trace)
                    .map_source(|_| one_device())
                    .build(),
            )
            .out_retry(Publish::default())
            .to("devices/dev42/retry");
            b.include(subscriber("fleet/firmware", ScheduleRollout).build());
        },
    )
}

fn main() -> ExitCode {
    run_main(app)
}

Фильтру соответствует много тем, поэтому у отложенной копии его доставки нет одной темы, куда её отправить. Такую тему называет точка монтирования через out_retry(Publish::default()).to(..), а без неё сервис не запускается. Копия доходит до каждой подписки, фильтр которой подходит к её теме, поэтому здесь к каждой такой теме подходит только свой фильтр. Копии разбирает руководство Переобработка и темы dead-letter.

Пересекающиеся фильтры

К devices/dev42/telemetry подходят и store, и trace, а соединение у них одно. На показание в этой теме каждый из них выполняется один раз. Сервер отправляет по копии на каждую подходящую подписку, а брокер различает копии по идентификатору подписки, указанному в каждой копии. Идентификаторы подписок сервер поддерживать не обязан. На сервере без них сервис не запускается и возвращает ошибку, в которой названы оба фильтра.

Один и тот же фильтр, смонтированный в сервисе дважды, для сервера - одна подписка. Его обработчики получают доставки по очереди, а не каждый свою копию.

Проверка маршрутизации

Тесты в файле примера запускают приложение внутри процесса. Они публикуют в devices/dev42/telemetry, devices/dev7/telemetry и devices/dev42/telemetry/raw и проверяют, какие обработчики выполнились. Внутрипроцессный сервер сопоставляет темы так же, как Mosquitto, включая подстановочные знаки и пересечения. Нескольким репликам в одной группе нужны несколько соединений с одним сервером, поэтому само деление видно только на настоящем брокере.