Запрос и ответ

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

Для этого MQTT 5 даёт запросу два свойства. Тема ответа указывает, куда уходит ответ. Данные корреляции - непрозрачное значение, по которому запрашивающая сторона сопоставляет ответ со своим вопросом. Брокер отображает их на заголовки reply-to и correlation-id в обе стороны.

Пример - агент устройства. Консоли запрашивают у него одну настройку в devices/dev42/config/get, а он отвечает её значением.

Отвечающая сторона

Макросы

src/main.rs

/// Asks the device for one setting.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "devices/dev42/config/get")]
struct ConfigQuery {
    key: String,
}

/// The answer. The type names no topic: each query says where its answer goes.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
struct ConfigValue {
    key: String,
    value: String,
}

/// Stands in for the device's settings.
fn setting(key: &str) -> Option<&'static str> {
    match key {
        "sample-interval" => Some("30s"),
        "firmware" => Some("2.1.0"),
        _ => None,
    }
}

/// A query without a response topic is answered on the topic the mount site names; a key the
/// device does not have gets no answer.
#[subscriber(
    MqttTopic::new("devices/dev42/config/get"),
    reply("devices/dev42/config")
)]
async fn answer(query: &ConfigQuery) -> Result<ConfigValue, HandlerOutcome> {
    let Some(value) = setting(&query.key) else {
        tracing::warn!(query.key, "unknown setting");
        return Err(HandlerOutcome::drop());
    };
    tracing::info!(query.key, "setting reported");
    Ok(ConfigValue {
        key: query.key.clone(),
        value: value.to_owned(),
    })
}

Вручную

src/main.rs

/// Asks the device for one setting.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
#[outgoing(name = "devices/dev42/config/get")]
struct ConfigQuery {
    key: String,
}

/// The answer. The type names no topic: each query says where its answer goes.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct ConfigValue {
    key: String,
    value: String,
}

/// Stands in for the device's settings.
fn setting(key: &str) -> Option<&'static str> {
    match key {
        "sample-interval" => Some("30s"),
        "firmware" => Some("2.1.0"),
        _ => None,
    }
}

/// A key the device does not have gets no answer.
struct Answer;

impl Handle<ConfigQuery, ConfigValue> for Answer {
    fn handle(
        &self,
        query: &ConfigQuery,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<ConfigValue, HandlerOutcome>> {
        let Some(value) = setting(&query.key) else {
            tracing::warn!(query.key, "unknown setting");
            return ready(Err(HandlerOutcome::drop()));
        };
        tracing::info!(query.key, "setting reported");
        ready(Ok(ConfigValue {
            key: query.key.clone(),
            value: value.to_owned(),
        }))
    }
}

Отвечающая сторона - это обычный обработчик с ответом: значение, которое он возвращает, публикуется. ConfigValue не называет темы, поэтому её называет точка монтирования: reply("devices/dev42/config"). Эта тема - запасная, для запроса без темы ответа. Запрос настройки, которой у устройства нет, заканчивается drop, и ничего не публикуется.

Ответ в тему ответа

Макросы

src/main.rs

/// Sends the answer to the response topic of the query, which arrives as the `reply-to` header,
/// and copies the query's correlation data, so the requester matches the answer to its question.
struct ToResponseTopic;

impl<C, Options> PublishTransform<ForReply<C>, Options> for ToResponseTopic {
    type Destination = Names;

    fn apply(
        &self,
        out: &mut Outgoing<'_>,
        _options: &mut Option<Options>,
        cx: &PublishContext<'_, C>,
    ) {
        if let Some(topic) = cx.headers().reply_to() {
            out.set_name(topic.to_owned());
        }
        if let Some(id) = cx.headers().correlation_id() {
            out.headers_mut().insert("correlation-id", id.to_owned());
        }
    }
}

#[ruststream::app]
fn app() -> impl App<State = ()> {
    RustStream::new(AppInfo::new("device-agent", "0.1.0")).with_broker(
        MqttBroker::new("mqtt://localhost:1883", "device-agent-dev42"),
        |b| {
            b.include(answer)
                .out_reply(Publish::default())
                .transform(ToResponseTopic);
        },
    )
}

Вручную

src/main.rs

/// Sends the answer to the response topic of the query, which arrives as the `reply-to` header,
/// and copies the query's correlation data, so the requester matches the answer to its question.
struct ToResponseTopic;

impl<C, Options> PublishTransform<ForReply<C>, Options> for ToResponseTopic {
    type Destination = Names;

    fn apply(
        &self,
        out: &mut Outgoing<'_>,
        _options: &mut Option<Options>,
        cx: &PublishContext<'_, C>,
    ) {
        if let Some(topic) = cx.headers().reply_to() {
            out.set_name(topic.to_owned());
        }
        if let Some(id) = cx.headers().correlation_id() {
            out.headers_mut().insert("correlation-id", id.to_owned());
        }
    }
}

fn app() -> impl App<State = ()> {
    RustStream::new(AppInfo::new("device-agent", "0.1.0")).with_broker(
        MqttBroker::new("mqtt://localhost:1883", "device-agent-dev42"),
        |b| {
            // A query without a response topic is answered on the topic named here.
            let answer = subscriber("devices/dev42/config/get", Answer)
                .reply()
                .to("devices/dev42/config")
                .build();
            b.include(answer)
                .out_reply(Publish::default())
                .transform(ToResponseTopic);
        },
    )
}

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

ToResponseTopic - преобразование публикации в позиции ответа. Оно читает заголовки запроса: reply-to становится темой ответа, а correlation-id копируется в ответ. Оно объявляет Destination = Names, и это даёт ему право выбирать тему. Запрос без reply-to оставляет тему точки монтирования.

Документ AsyncAPI сервиса указывает адресом ответа заголовок reply-to запроса, $message.header#/reply-to, потому что ответы не собираются ни в одном фиксированном канале.

Запрос из командной строки

Запрашивающей стороной может быть любой клиент MQTT 5. В Mosquitto входит mosquitto_rr: он публикует запрос с темой ответа и печатает ответ:

docker compose exec mosquitto mosquitto_rr \
  -t devices/dev42/config/get -e consoles/c1/responses -m '{"key":"firmware"}'

Он печатает {"key":"firmware","value":"2.1.0"}. Запрашивающая сторона внутри сервиса на RustStream публикует запрос с обоими заголовками и читает ответы подписчиком на собственную тему ответа.

Проверка отвечающей стороны

Тесты в файле примера публикуют на внутрипроцессном сервере запрос с заголовками reply-to и correlation-id. Заголовки доходят до обработчика через то же отображение на свойства MQTT 5, что и при передаче по сети. Тесты проверяют ответ в consoles/c1/responses с данными корреляции, запасную тему для запроса без темы ответа и то, что на неизвестную настройку ответа нет.