Responders

A responder answers on the response topic a request carries, through a reply transform that also copies the correlation data.

An MQTT 5 request carries the topic its answer goes to in the response topic property, and the handler reads it as the reply-to header. Its correlation data arrives as the correlation-id header, and the requester matches the answer to its question by it. A PublishTransform on the reply position reads both from the request: cx.headers().reply_to() and cx.headers().correlation_id(). It names the answer's topic, so it declares type Destination = Names. The reply type declares no topic of its own, which is what leaves the naming to the transform. The topic the mount site names, reply("..") in the macro form, is the fallback: a request without a response topic is answered there. The generated document points the reply's address at the request's reply-to header, as Replies shows.

Example

A device agent answers each status request on the response topic of the console that asked:

Macros
Manual

examples/mqtt_publishing_responders.rs
/// Answers on the response topic of the request, which arrives as the `reply-to` header, and
/// copies the request'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());
        }
    }
}

/// A request without a response topic is answered on the topic the mount site names.
#[subscriber(
    MqttTopic::new("devices/dev42/status/get"),
    reply("devices/dev42/status")
)]
async fn report(request: &StatusRequest) -> Status {
    tracing::info!(request.detail, "status requested");
    Status {
        firmware: "1.4.2".to_owned(),
        online: true,
    }
}

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

examples/manual/mqtt_publishing_responders.rs
/// Answers on the response topic of the request, which arrives as the `reply-to` header, and
/// copies the request'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());
        }
    }
}

struct Report;

impl Handle<StatusRequest, Status> for Report {
    fn handle(
        &self,
        request: &StatusRequest,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Status, HandlerOutcome>> {
        tracing::info!(request.detail, "status requested");
        ready(Ok(Status {
            firmware: "1.4.2".to_owned(),
            online: true,
        }))
    }
}

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

Example​

Example