Публикация команд

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

Пример - сервис-диспетчер команд. Операторы отправляют запрос команды в operators/commands. Сервис пересылает каждую команду в тему её устройства, devices/<id>/commands.

Тема для каждого устройства

Макросы

src/main.rs

/// What a device reads on its own command topic; the `{device}` level becomes a setter.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "devices/{device}/commands")]
struct DeviceCommand {
    action: String,
}

#[derive(OutSlot)]
#[publishes(DeviceCommand)]
struct Commands;

Вручную

src/main.rs

/// What a device reads on its own command topic; the `{device}` level becomes a setter.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
#[outgoing(name = "devices/{device}/commands")]
struct DeviceCommand {
    action: String,
}

#[derive(OutSlot)]
#[publishes(DeviceCommand)]
struct Commands;

Тема DeviceCommand - шаблон имени. {device} становится сеттером публикации, поэтому обработчик называет устройство, а тема строится из него. Commands - слот, через который публикует обработчик, и публикует он только DeviceCommand.

Команда

Макросы

src/main.rs

/// Forwards a command to its device. An urgent one goes out at `QoS` 2, every other one at the
/// quality of service of the slot's policy; the operator travels as a header.
#[subscriber("operators/commands")]
async fn dispatch(
    request: &CommandRequest,
    Out(commands): Out<impl Publisher<Options = MqttPublishOptions>, Commands>,
) -> HandlerOutcome {
    let mut headers = HeaderMap::new();
    headers.insert("operator", request.operator.clone());
    let command = DeviceCommand {
        action: request.action.clone(),
    };
    let message = commands.message(&command).with_headers(headers);
    let message = if request.urgent {
        message.qos(Qos::ExactlyOnce)
    } else {
        message
    };
    if let Err(err) = message.to().device(&request.device).publish().await {
        tracing::warn!(request.device, %err, "command refused");
        return HandlerOutcome::drop();
    }
    tracing::info!(request.device, request.action, "command sent");
    HandlerOutcome::ack()
}

Вручную

src/main.rs

/// Forwards a command to its device. An urgent one goes out at `QoS` 2, every other one at the
/// quality of service of the slot's policy; the operator travels as a header.
struct Dispatch;

impl<Entry> Handle<CommandRequest, (), Outs<(Entry,)>> for Dispatch
where
    Entry: OutEntry<Commands, Wire: Publisher<Options = MqttPublishOptions>>,
{
    async fn handle(
        &self,
        request: &CommandRequest,
        outs: &Outs<(Entry,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let mut headers = HeaderMap::new();
        headers.insert("operator", request.operator.clone());
        let command = DeviceCommand {
            action: request.action.clone(),
        };
        let message = outs.get(Commands).message(&command).with_headers(headers);
        let message = if request.urgent {
            message.qos(Qos::ExactlyOnce)
        } else {
            message
        };
        if let Err(err) = message.to().device(&request.device).publish().await {
            tracing::warn!(request.device, %err, "command refused");
            return Err(HandlerOutcome::drop());
        }
        tracing::info!(request.device, request.action, "command sent");
        Ok(())
    }
}

Политика слота задаёт качество обслуживания каждой команды. Для срочной команды вызывается шаг qos, и она уходит с QoS 2. Шаг меняет одно сообщение. Настройку, которую вызов не называет, берёт политика. Шаг записывает MqttPublishOptions, поэтому ограничение слота называет этот тип, а файл импортирует прелюдию брокера. Второй шаг - retain, и руководство Состояние устройств задаёт его в политике.

Заголовок operator передаётся пользовательским свойством MQTT 5. Устройство читает его как обычное свойство, без конверта вокруг полезной нагрузки. Заголовки content-type, reply-to и correlation-id вместо этого занимают собственные свойства протокола: тип содержимого, тему ответа и данные корреляции.

publish возвращает управление, как только сообщение принял клиент, до подтверждения от сервера. При QoS 1 и 2 клиент отправляет сообщение повторно, пока сервер его не подтвердит, в том числе после переподключений. Ошибка из publish означает, что клиент не принял сообщение: соединение закрыто или сообщение нарушает правила протокола. Тема с + или # их нарушает: запрос называет устройство +, тема становится devices/+/commands, и dispatch отбрасывает запрос. Заголовок или тема с управляющим символом отклоняются так же, потому что сервер может закрыть соединение из-за такого символа.

Тип содержимого

Макросы

src/main.rs

/// States the media type of everything the slot sends, which fills the content type property
/// and tells a device to read the body as UTF-8 text.
struct Json;

impl<K: ContextKind, Options> PublishTransform<K, Options> for Json {
    type Destination = Reads;

    fn apply(&self, out: &mut Outgoing<'_>, _options: &mut Option<Options>, _cx: &K::View<'_>) {
        out.headers_mut().insert("content-type", "application/json");
    }
}

#[ruststream::app]
fn app() -> impl App<State = ()> {
    RustStream::new(AppInfo::new("commander", "0.1.0")).with_broker(
        MqttBroker::new("mqtt://localhost:1883", "commander"),
        |b| {
            b.include(dispatch)
                .out(Commands, Publish::default().qos(Qos::AtLeastOnce))
                .transform(Json)
                .build();
        },
    )
}

Вручную

src/main.rs

/// States the media type of everything the slot sends, which fills the content type property
/// and tells a device to read the body as UTF-8 text.
struct Json;

impl<K: ContextKind, Options> PublishTransform<K, Options> for Json {
    type Destination = Reads;

    fn apply(&self, out: &mut Outgoing<'_>, _options: &mut Option<Options>, _cx: &K::View<'_>) {
        out.headers_mut().insert("content-type", "application/json");
    }
}

fn app() -> impl App<State = ()> {
    RustStream::new(AppInfo::new("commander", "0.1.0")).with_broker(
        MqttBroker::new("mqtt://localhost:1883", "commander"),
        |b| {
            b.include(subscriber("operators/commands", Dispatch).build())
                .out(Commands, Publish::default().qos(Qos::AtLeastOnce))
                .transform(Json)
                .build();
        },
    )
}

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

Точка монтирования задаёт слоту политику с QoS 1 и преобразование публикации. Json ставит content-type: application/json на каждую команду, которую отправляет слот. Брокер превращает заголовок в свойство типа содержимого и выставляет индикатор формата полезной нагрузки, поэтому устройство читает тело как текст в UTF-8. Сам кодек этот заголовок не пишет: тип содержимого сервис указывает один раз, преобразованием в точке монтирования.

Проверка команд

Тесты в файле примера запускают приложение внутри процесса и читают команды, которые опубликовал слот. tb.out::<Commands>() показывает настройки каждого сообщения: никаких у обычной команды и QoS 2 у срочной. published в devices/dev42/commands показывает полезную нагрузку и оба заголовка. Внутрипроцессный сервер проверяет темы и заголовки так же, как клиент, поэтому запрос для устройства + отбрасывается и там.