Начало работы

Это руководство устанавливает ruststream-rumqttc, запускает локальный сервер MQTT и первый сервис на нём.

Установка

Cargo.toml

[dependencies]
ruststream = { version = "0.7", features = ["macros", "json", "logging"] }
ruststream-rumqttc = "0.7"
serde = { version = "1", features = ["derive"] }
tracing = "0.1"

[dev-dependencies]
ruststream-rumqttc = { version = "0.7", features = ["testing"] }
tokio = { version = "1", features = ["macros", "rt"] }

ruststream - это сам фреймворк. Фича macros даёт атрибуты, json - кодек, а logging выводит журнал сервиса в терминал. ruststream-rumqttc - брокер MQTT 5. Собственного кодека брокер не подключает, поэтому сервис выбирает json, msgpack или cbor в фичах ruststream.

У крейта брокера две фичи, обе выключены по умолчанию:

Фича	Что она даёт
testing	внутрипроцессный сервер MQTT, на котором работает TestApp
asyncapi	привязки MQTT в генерируемом документе AsyncAPI

Системные требования

Клиент rumqttc работает с TLS через rustls, а криптографию rustls берёт из aws-lc-rs. aws-lc-rs компилирует код на C, поэтому для сборки нужен компилятор C. В Debian и Ubuntu его ставит пакет build-essential. Требования для других платформ перечислены в документации aws-lc-rs.

Локальный сервер MQTT

Для разработки достаточно одного контейнера Eclipse Mosquitto. Свои слушатели он читает из файла конфигурации:

mosquitto.conf

listener 1883
allow_anonymous true

compose.yaml

services:
  mosquitto:
    image: eclipse-mosquitto:2
    ports:
      - "1883:1883"
    volumes:
      - ./mosquitto.conf:/mosquitto/config/mosquitto.conf:ro

docker compose up -d

Сервер слушает localhost:1883 и принимает клиентов без пароля.

Первый сервис

Макросы

src/main.rs

/// The coldest temperature a sensor can report; anything below it is a broken sensor.
const ABSOLUTE_ZERO: f64 = -273.15;

#[derive(Debug, PartialEq, Deserialize)]
struct Telemetry {
    device: String,
    temperature: f64,
}

#[subscriber(MqttFilter::new("devices/+/telemetry").qos(Qos::AtLeastOnce))]
async fn record(telemetry: &Telemetry) -> HandlerOutcome {
    if telemetry.temperature < ABSOLUTE_ZERO {
        tracing::warn!(telemetry.device, "impossible reading dropped");
        return HandlerOutcome::drop();
    }
    tracing::info!(telemetry.device, telemetry.temperature, "reading recorded");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("telemetry", "0.1.0")).with_broker(
        MqttBroker::new("mqtt://localhost:1883", "telemetry-svc"),
        |b| {
            b.include(record)
                .out_retry(Publish::default())
                .to("devices/retry/telemetry");
        },
    )
}

Вручную

src/main.rs

/// The coldest temperature a sensor can report; anything below it is a broken sensor.
const ABSOLUTE_ZERO: f64 = -273.15;

#[derive(Debug, PartialEq, Deserialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Telemetry {
    device: String,
    temperature: f64,
}

struct Record;

impl Handle<Telemetry> for Record {
    fn handle(
        &self,
        telemetry: &Telemetry,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        if telemetry.temperature < ABSOLUTE_ZERO {
            tracing::warn!(telemetry.device, "impossible reading dropped");
            return ready(Err(HandlerOutcome::drop()));
        }
        tracing::info!(telemetry.device, telemetry.temperature, "reading recorded");
        ready(Ok(()))
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("telemetry", "0.1.0")).with_broker(
        MqttBroker::new("mqtt://localhost:1883", "telemetry-svc"),
        |b| {
            let fleet = MqttFilter::new("devices/+/telemetry").qos(Qos::AtLeastOnce);
            b.include(
                subscriber("devices/+/telemetry", Record)
                    .map_source(|_| fleet)
                    .build(),
            )
            .out_retry(Publish::default())
            .to("devices/retry/telemetry");
        },
    )
}

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

record читает телеметрию каждого устройства в JSON. Показание он подтверждает, а показание ниже абсолютного нуля отбрасывает. MqttFilter подписывается на фильтр темы: + заменяет один уровень темы, здесь идентификатор устройства. Qos::AtLeastOnce делает доставку каждого показания подтверждаемой: подтверждение уходит серверу, когда record возвращает управление. Фильтру соответствует много тем, поэтому тему для отложенной переобработки называет точка монтирования. Её называет каждая подписка на фильтр, а без неё сервис не запускается.

MqttBroker::new принимает URL сервера и идентификатор клиента и только запоминает их. Соединение открывается при запуске сервиса. По идентификатору клиента сервер узнаёт сессию сервиса, поэтому у каждого запущенного процесса он свой. #[ruststream::app] превращает app в main. В ручной форме то же делает run_main.

Запустите сервис:

cargo run -- run

Опубликуйте показание из другого терминала клиентом, который входит в образ:

docker compose exec mosquitto mosquitto_pub -V mqttv5 -q 1 \
  -t devices/dev42/telemetry -m '{"device":"dev42","temperature":21.5}'

Сервис пишет в журнал reading recorded с устройством и его температурой. Ctrl+C останавливает его.

Первый тест

В файле примера есть и два теста. Каждый запускает то же app через TestApp::start, публикует показание в devices/dev42/telemetry и проверяет, как record завершил его доставку. Там MqttBroker подключается к внутрипроцессному серверу MQTT, поэтому тестам не нужен Mosquitto:

cargo test

Руководство Тестирование сервиса на MQTT разбирает эти тесты и то, что моделирует внутрипроцессный сервер.

Переход с rumqttc

Сервис на rumqttc сам опрашивает цикл событий и вручную подтверждает каждое сообщение. Вот куда переходит каждая часть в RustStream:

rumqttc	RustStream
MqttOptions и его сеттеры	билдер MqttBroker: keep_alive, credentials, clean_start, last_will
цикл EventLoop::poll	собственная задача соединения брокера, которая переподключается и подписывается заново
AsyncClient::subscribe	обработчик #[subscriber] на MqttTopic или MqttFilter
десериализация publish.payload	тип параметра обработчика, который декодирует кодек
set_manual_acks(true) и AsyncClient::ack	исход обработчика (Гарантии доставки)
AsyncClient::publish	ответ или параметр Out, отправленный через политику публикации (Публикация команд)