ruststream_rumqttc/
message.rs

1//! [`MqttMessage`] and the mapping between `RustStream` headers and MQTT 5 properties.
2//!
3//! User properties carry headers natively; the well-known `content-type`, `reply-to`, and
4//! `correlation-id` headers ride the matching first-class MQTT 5 properties, so no envelope
5//! format is invented and non-Rust peers see plain MQTT messages.
6
7use bytes::Bytes;
8use rumqttc::v5::AsyncClient;
9use rumqttc::v5::mqttbytes::QoS;
10use rumqttc::v5::mqttbytes::v5::{Publish, PublishProperties};
11use ruststream::{AckError, Headers, IncomingMessage, OutgoingMessage};
12
13/// A message delivered by an [`MqttSubscriber`](crate::MqttSubscriber).
14///
15/// `ack` acknowledges through the protocol for `QoS` 1 (`PUBACK`) and `QoS` 2 (`PUBREC`, with the
16/// client completing the handshake); `QoS` 0 deliveries report
17/// [`AckError::Unsupported`]. MQTT has no negative acknowledgement, so `nack(requeue = true)`
18/// reports [`AckError::Unsupported`] as well - unacknowledged messages redeliver when the
19/// session resumes - and `nack(requeue = false)` acknowledges (dropping is the only terminal
20/// outcome the protocol offers).
21pub struct MqttMessage {
22    payload: Bytes,
23    headers: Headers,
24    topic: String,
25    /// `None` when this delivery carries no acknowledgement: `QoS` 0, or a fanned-out copy on
26    /// an overlapping filter (the wire ack belongs to exactly one delivery).
27    acker: Option<(AsyncClient, Publish)>,
28}
29
30impl std::fmt::Debug for MqttMessage {
31    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
32        f.debug_struct("MqttMessage")
33            .field("topic", &self.topic)
34            .field("payload_len", &self.payload.len())
35            .finish_non_exhaustive()
36    }
37}
38
39impl MqttMessage {
40    pub(crate) fn new(topic: String, publish: &Publish, client: Option<AsyncClient>) -> Self {
41        let mut headers = Headers::new();
42        if let Some(properties) = &publish.properties {
43            for (name, value) in &properties.user_properties {
44                headers.insert(name.clone(), value.clone());
45            }
46            if let Some(content_type) = &properties.content_type {
47                headers.insert("content-type", content_type.clone());
48            }
49            if let Some(response_topic) = &properties.response_topic {
50                headers.insert("reply-to", response_topic.clone());
51            }
52            if let Some(correlation) = &properties.correlation_data {
53                headers.insert("correlation-id", correlation.clone());
54            }
55        }
56        let acker = match publish.qos {
57            QoS::AtMostOnce => None,
58            _ => client.map(|client| (client, publish.clone())),
59        };
60        Self {
61            payload: publish.payload.clone(),
62            headers,
63            topic,
64            acker,
65        }
66    }
67
68    /// The topic this message was published to (the real topic, never a `$share` filter).
69    #[must_use]
70    pub fn topic(&self) -> &str {
71        &self.topic
72    }
73}
74
75impl IncomingMessage for MqttMessage {
76    fn payload(&self) -> &[u8] {
77        &self.payload
78    }
79
80    fn headers(&self) -> &Headers {
81        &self.headers
82    }
83
84    async fn ack(self) -> Result<(), AckError> {
85        let Some((client, publish)) = self.acker else {
86            return Err(AckError::Unsupported);
87        };
88        client
89            .ack(&publish)
90            .await
91            .map_err(|_| AckError::Broker(Box::from("the mqtt connection task has shut down")))
92    }
93
94    async fn nack(self, requeue: bool) -> Result<(), AckError> {
95        if requeue {
96            // MQTT has no negative acknowledgement: an unacked message redelivers only when
97            // the session resumes. Reporting Unsupported is honest; pretending would ack.
98            Err(AckError::Unsupported)
99        } else {
100            self.ack().await
101        }
102    }
103}
104
105/// Builds the wire properties for an outgoing publish. Returns `None` when the message
106/// carries no headers, so plain messages stay property-free on the wire.
107pub(crate) fn to_publish_properties(msg: &OutgoingMessage<'_>) -> Option<PublishProperties> {
108    let headers = msg.headers();
109    if headers.is_empty() {
110        return None;
111    }
112    let mut properties = PublishProperties::default();
113    for (name, value) in headers.iter() {
114        let text = String::from_utf8_lossy(value).into_owned();
115        match name {
116            "content-type" => properties.content_type = Some(text),
117            "reply-to" => properties.response_topic = Some(text),
118            "correlation-id" => {
119                properties.correlation_data = Some(Bytes::copy_from_slice(value));
120            }
121            other => properties.user_properties.push((other.to_owned(), text)),
122        }
123    }
124    Some(properties)
125}
126
127#[cfg(test)]
128mod tests {
129    use super::*;
130
131    #[test]
132    fn well_known_headers_ride_first_class_properties() {
133        let mut headers = Headers::new();
134        headers.insert("content-type", "application/json");
135        headers.insert("reply-to", "replies/1");
136        headers.insert("correlation-id", "corr-1");
137        headers.insert("x-tenant", "acme");
138        let outgoing = OutgoingMessage::new("orders", b"{}".as_slice()).with_headers(headers);
139
140        let properties = to_publish_properties(&outgoing).expect("properties built");
141        assert_eq!(properties.content_type.as_deref(), Some("application/json"));
142        assert_eq!(properties.response_topic.as_deref(), Some("replies/1"));
143        assert_eq!(
144            properties.correlation_data.as_deref(),
145            Some(b"corr-1".as_slice())
146        );
147        assert_eq!(
148            properties.user_properties,
149            vec![("x-tenant".to_owned(), "acme".to_owned())]
150        );
151    }
152
153    #[test]
154    fn plain_messages_stay_property_free() {
155        let outgoing = OutgoingMessage::new("orders", b"{}".as_slice());
156        assert!(to_publish_properties(&outgoing).is_none());
157    }
158}
ruststream_rumqttc/message.rs

ruststream_rumqttc/
message.rs
