Headers

Headers travel as MQTT 5 user properties; content-type, reply-to and correlation-id take the protocol's own properties.

A peer in any language reads them as plain MQTT, with no envelope around the payload. content-type becomes the content type property, reply-to the response topic a responder answers on, and correlation-id the correlation data, in both directions. A message with no headers is published with no properties at all. A textual content type (application/json, any text/ type, any +json type) sets the payload format indicator to 1, which tells a peer the body is UTF-8; any other sets it to 0. A property is an MQTT string, and a server may close the connection on one with a control character. So a header name or value that is not UTF-8, or carries a control character or a Unicode non-character, refuses the publish with MqttError::Publish. The codec of a position writes no header, so nothing sets the content type unless the service does. A publish transform sets it at the mount site, once for every message of one position: the replies, a slot or the retry copies.

Example

Every reply of the position states that its body is JSON:

Macros
Manual

examples/mqtt_publishing_headers.rs
/// States the media type of every message the position sends. The header fills the content type
/// property and sets the payload format indicator, so a peer reads the body as UTF-8 JSON. The
/// transform reads nothing of the delivery, so it mounts on any position.
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");
    }
}

#[subscriber(MqttTopic::new("devices/dev42/telemetry"), reply)]
async fn track(telemetry: &Telemetry) -> DeviceState {
    DeviceState {
        hot: telemetry.temperature > 30.0,
    }
}

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

examples/manual/mqtt_publishing_headers.rs
/// States the media type of every message the position sends. The header fills the content type
/// property and sets the payload format indicator, so a peer reads the body as UTF-8 JSON. The
/// transform reads nothing of the delivery, so it mounts on any position.
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");
    }
}

struct Track;

impl Handle<Telemetry, DeviceState> for Track {
    fn handle(
        &self,
        telemetry: &Telemetry,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<DeviceState, HandlerOutcome>> {
        ready(Ok(DeviceState {
            hot: telemetry.temperature > 30.0,
        }))
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("states", "0.1.0")).with_broker(
        MqttBroker::new("mqtt://localhost:1883", "states-svc"),
        |b| {
            b.include(subscriber("devices/dev42/telemetry", Track).reply().build())
                .out_reply(Publish::default())
                .transform(Json);
        },
    )
}

Example​

Example