Skip to main content

Request and response

This guide answers MQTT 5 requests: each answer goes to the topic its request names and carries the request's correlation data.

MQTT 5 gives a request two properties for this. The response topic says where the answer goes. The correlation data is an opaque value the requester uses to match an answer to its question. The broker maps them onto the reply-to and correlation-id headers in both directions.

The example is a device agent. Consoles ask it for one setting on devices/dev42/config/get, and it answers with the setting's value.

The responder​

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(),
    })
}

A responder is an ordinary handler with a reply: the value it returns is published. ConfigValue names no topic, so the mount site names one: reply("devices/dev42/config"). That topic is the fallback, for a request that carries no response topic. A setting the device does not have ends in drop, and nothing is published.

Answering on the response topic​

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);
        },
    )
}

ToResponseTopic is a publish transform on the reply position. It reads the request's headers: reply-to becomes the answer's topic, and correlation-id is copied onto the answer. It declares Destination = Names, which gives it the right to choose the topic. A request without reply-to keeps the mount site's topic.

The AsyncAPI document of the service reports the reply's address as the request's reply-to header, $message.header#/reply-to, because no fixed channel holds the answers.

Asking from the command line​

The requester is any MQTT 5 client. Mosquitto ships one, mosquitto_rr, which publishes a request with a response topic and prints the answer:

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

It prints {"key":"firmware","value":"2.1.0"}. A requester inside a RustStream service publishes the request with both headers set, and reads the answers with a subscriber on its own response topic.

Testing the responder​

The tests in the example file publish a query with reply-to and correlation-id headers, in process. The headers reach the handler through the same mapping onto MQTT 5 properties as on the wire. The tests check the answer on consoles/c1/responses with its correlation data, the fallback topic for a query without a response topic, and that an unknown setting gets no answer.