pamoja_mqtt/lib.rs
1//! MQTT transport for the pamoja SDK.
2//!
3//! [`MqttTransport`] implements the core [`Transport`]
4//! trait on top of the pure-Rust [`rumqttc`] client, so an application can publish
5//! to and subscribe from an MQTT broker through the same protocol-agnostic surface
6//! it uses for every other transport.
7//!
8//! Once [`connect`](Transport::connect) succeeds the transport owns a background
9//! task that drives the MQTT event loop: it answers keep-alive pings, completes
10//! delivery handshakes, and forwards inbound messages to an internal queue that
11//! [`recv`](MqttTransport::recv) drains. Publishing and subscribing use the
12//! default [`QualityOfService`] configured on the transport.
13//!
14//! # Examples
15//!
16//! ```no_run
17//! use pamoja_core::Transport;
18//! use pamoja_mqtt::{MqttConfig, MqttTransport};
19//!
20//! # async fn run() -> pamoja_core::Result<()> {
21//! let mut transport = MqttTransport::new(MqttConfig::new("sensor-1", "localhost", 1883));
22//! transport.connect().await?;
23//! transport.subscribe("sensors/+/temperature").await?;
24//! transport.send("sensors/1/temperature", b"21.5").await?;
25//!
26//! if let Some(message) = transport.recv().await? {
27//! println!("{}: {} bytes", message.topic, message.payload.len());
28//! }
29//! # Ok(())
30//! # }
31//! ```
32
33use std::time::Duration;
34
35use pamoja_core::{Error, Result, Transport};
36use rumqttc::{AsyncClient, ClientError, ConnectionError, Event, MqttOptions, Packet, QoS};
37use tokio::sync::{mpsc, oneshot};
38use tokio::task::JoinHandle;
39
40/// The delivery guarantee applied to published and subscribed messages.
41///
42/// These map one-to-one onto the MQTT protocol's quality-of-service levels.
43#[derive(Clone, Copy, Debug, PartialEq, Eq)]
44pub enum QualityOfService {
45 /// Fire and forget: the broker does not acknowledge delivery.
46 AtMostOnce,
47 /// The message is delivered at least once and acknowledged.
48 AtLeastOnce,
49 /// The message is delivered exactly once via a four-step handshake.
50 ExactlyOnce,
51}
52
53impl From<QualityOfService> for QoS {
54 fn from(value: QualityOfService) -> Self {
55 match value {
56 QualityOfService::AtMostOnce => QoS::AtMostOnce,
57 QualityOfService::AtLeastOnce => QoS::AtLeastOnce,
58 QualityOfService::ExactlyOnce => QoS::ExactlyOnce,
59 }
60 }
61}
62
63/// Connection settings for an [`MqttTransport`].
64///
65/// Construct with [`MqttConfig::new`] and refine with the chained setters; every
66/// field has a sensible default so only the broker address and client id are
67/// required.
68#[derive(Clone, Debug)]
69pub struct MqttConfig {
70 client_id: String,
71 host: String,
72 port: u16,
73 keep_alive: Duration,
74 capacity: usize,
75 qos: QualityOfService,
76}
77
78impl MqttConfig {
79 /// Creates a configuration for the given client id and broker address.
80 ///
81 /// # Arguments
82 ///
83 /// * `client_id` - the MQTT client identifier presented to the broker.
84 /// * `host` - the broker hostname or IP address.
85 /// * `port` - the broker TCP port, conventionally `1883` for plaintext MQTT.
86 ///
87 /// # Returns
88 ///
89 /// A configuration with a 30-second keep-alive, a request capacity of 64, and
90 /// a default quality of service of [`QualityOfService::AtLeastOnce`].
91 pub fn new(client_id: impl Into<String>, host: impl Into<String>, port: u16) -> Self {
92 Self {
93 client_id: client_id.into(),
94 host: host.into(),
95 port,
96 keep_alive: Duration::from_secs(30),
97 capacity: 64,
98 qos: QualityOfService::AtLeastOnce,
99 }
100 }
101
102 /// Sets the keep-alive interval used to hold the connection open.
103 ///
104 /// # Arguments
105 ///
106 /// * `interval` - how often the client pings the broker when otherwise idle.
107 ///
108 /// # Returns
109 ///
110 /// The updated configuration, for chaining.
111 pub fn keep_alive(mut self, interval: Duration) -> Self {
112 self.keep_alive = interval;
113 self
114 }
115
116 /// Sets the bound on outstanding client requests buffered toward the broker.
117 ///
118 /// # Arguments
119 ///
120 /// * `capacity` - the request channel capacity; values below one are clamped
121 /// to one.
122 ///
123 /// # Returns
124 ///
125 /// The updated configuration, for chaining.
126 pub fn capacity(mut self, capacity: usize) -> Self {
127 self.capacity = capacity.max(1);
128 self
129 }
130
131 /// Sets the default quality of service for publishes and subscriptions.
132 ///
133 /// # Arguments
134 ///
135 /// * `qos` - the delivery guarantee applied by [`send`](Transport::send) and
136 /// [`subscribe`](Transport::subscribe).
137 ///
138 /// # Returns
139 ///
140 /// The updated configuration, for chaining.
141 pub fn qos(mut self, qos: QualityOfService) -> Self {
142 self.qos = qos;
143 self
144 }
145}
146
147/// A message received from a subscribed topic.
148#[derive(Clone, Debug, PartialEq, Eq)]
149pub struct Message {
150 /// The topic the message was published to.
151 pub topic: String,
152 /// The raw payload bytes.
153 pub payload: Vec<u8>,
154}
155
156/// An MQTT client that implements the core [`Transport`] trait.
157///
158/// A transport is created disconnected; [`connect`](Transport::connect) opens the
159/// link and spawns the background task that runs the MQTT event loop for the life
160/// of the connection. Inbound messages are queued and read with
161/// [`recv`](MqttTransport::recv).
162pub struct MqttTransport {
163 config: MqttConfig,
164 client: Option<AsyncClient>,
165 incoming: Option<mpsc::UnboundedReceiver<Message>>,
166 pump: Option<JoinHandle<()>>,
167}
168
169impl MqttTransport {
170 /// Creates a transport from the given configuration without connecting.
171 ///
172 /// # Arguments
173 ///
174 /// * `config` - the broker connection settings.
175 ///
176 /// # Returns
177 ///
178 /// A disconnected transport ready for [`connect`](Transport::connect).
179 pub fn new(config: MqttConfig) -> Self {
180 Self {
181 config,
182 client: None,
183 incoming: None,
184 pump: None,
185 }
186 }
187
188 /// Reports whether the transport currently holds an active connection.
189 ///
190 /// # Returns
191 ///
192 /// `true` once [`connect`](Transport::connect) has succeeded and before
193 /// [`disconnect`](MqttTransport::disconnect) is called.
194 pub fn is_connected(&self) -> bool {
195 self.client.is_some()
196 }
197
198 /// Awaits the next message from any subscribed topic.
199 ///
200 /// # Returns
201 ///
202 /// `Some(message)` for the next queued message, or `None` once the event loop
203 /// has stopped and no further messages will arrive.
204 ///
205 /// # Errors
206 ///
207 /// Returns [`Error::Closed`] if the transport
208 /// is not connected.
209 pub async fn recv(&mut self) -> Result<Option<Message>> {
210 let incoming = self.incoming.as_mut().ok_or(Error::Closed)?;
211 Ok(incoming.recv().await)
212 }
213
214 /// Closes the connection and stops the background event loop.
215 ///
216 /// Calling this on a transport that is not connected is a no-op.
217 ///
218 /// # Returns
219 ///
220 /// `Ok(())` once the disconnect request has been issued and the event loop
221 /// task has been stopped.
222 ///
223 /// # Errors
224 ///
225 /// This call is best-effort and does not surface broker errors raised while
226 /// tearing down, so it currently always returns `Ok(())`.
227 pub async fn disconnect(&mut self) -> Result<()> {
228 if let Some(client) = self.client.take() {
229 let _ = client.disconnect().await;
230 }
231 if let Some(pump) = self.pump.take() {
232 pump.abort();
233 }
234 self.incoming = None;
235 Ok(())
236 }
237}
238
239impl Transport for MqttTransport {
240 async fn connect(&mut self) -> Result<()> {
241 let mut options = MqttOptions::new(
242 self.config.client_id.clone(),
243 self.config.host.clone(),
244 self.config.port,
245 );
246 options.set_keep_alive(self.config.keep_alive);
247
248 let (client, mut eventloop) = AsyncClient::new(options, self.config.capacity);
249 let (tx, rx) = mpsc::unbounded_channel();
250 let (ready_tx, ready_rx) = oneshot::channel::<Result<()>>();
251
252 let pump = tokio::spawn(async move {
253 let mut ready_tx = Some(ready_tx);
254 loop {
255 match eventloop.poll().await {
256 Ok(Event::Incoming(Packet::ConnAck(_))) => {
257 if let Some(ready_tx) = ready_tx.take() {
258 let _ = ready_tx.send(Ok(()));
259 }
260 }
261 Ok(Event::Incoming(Packet::Publish(publish))) => {
262 let message = Message {
263 topic: publish.topic,
264 payload: publish.payload.to_vec(),
265 };
266 if tx.send(message).is_err() {
267 break;
268 }
269 }
270 Ok(_) => {}
271 Err(err) => {
272 if let Some(ready_tx) = ready_tx.take() {
273 let _ = ready_tx.send(Err(map_connection_error(err)));
274 }
275 break;
276 }
277 }
278 }
279 });
280
281 match ready_rx.await {
282 Ok(Ok(())) => {
283 self.client = Some(client);
284 self.incoming = Some(rx);
285 self.pump = Some(pump);
286 Ok(())
287 }
288 Ok(Err(err)) => {
289 pump.abort();
290 Err(err)
291 }
292 Err(_) => {
293 pump.abort();
294 Err(Error::Transport(
295 "event loop closed before the connection was established".into(),
296 ))
297 }
298 }
299 }
300
301 async fn send(&mut self, topic: &str, payload: &[u8]) -> Result<()> {
302 let client = self.client.as_ref().ok_or(Error::Closed)?;
303 client
304 .publish(topic, self.config.qos.into(), false, payload.to_vec())
305 .await
306 .map_err(map_client_error)
307 }
308
309 async fn subscribe(&mut self, topic: &str) -> Result<()> {
310 let client = self.client.as_ref().ok_or(Error::Closed)?;
311 client
312 .subscribe(topic, self.config.qos.into())
313 .await
314 .map_err(map_client_error)
315 }
316}
317
318/// Maps a `rumqttc` client error onto the shared transport error.
319fn map_client_error(err: ClientError) -> Error {
320 Error::Transport(err.to_string())
321}
322
323/// Maps a `rumqttc` event-loop error onto the shared transport error.
324fn map_connection_error(err: ConnectionError) -> Error {
325 Error::Transport(err.to_string())
326}
327
328#[cfg(test)]
329mod tests {
330 use super::*;
331 use std::net::TcpListener;
332
333 /// Returns a TCP port with no listener bound, for negative connection tests.
334 fn unused_port() -> u16 {
335 let listener = TcpListener::bind("127.0.0.1:0").expect("bind ephemeral port");
336 listener.local_addr().expect("local addr").port()
337 }
338
339 #[test]
340 fn quality_of_service_maps_to_rumqttc() {
341 assert_eq!(QoS::from(QualityOfService::AtMostOnce), QoS::AtMostOnce);
342 assert_eq!(QoS::from(QualityOfService::AtLeastOnce), QoS::AtLeastOnce);
343 assert_eq!(QoS::from(QualityOfService::ExactlyOnce), QoS::ExactlyOnce);
344 }
345
346 #[test]
347 fn capacity_is_clamped_to_at_least_one() {
348 let config = MqttConfig::new("c", "localhost", 1883).capacity(0);
349 assert_eq!(config.capacity, 1);
350 }
351
352 #[tokio::test]
353 async fn send_before_connect_reports_closed() {
354 let mut transport = MqttTransport::new(MqttConfig::new("c", "localhost", 1883));
355 assert!(matches!(
356 transport.send("t", b"x").await,
357 Err(Error::Closed)
358 ));
359 }
360
361 #[tokio::test]
362 async fn subscribe_before_connect_reports_closed() {
363 let mut transport = MqttTransport::new(MqttConfig::new("c", "localhost", 1883));
364 assert!(matches!(transport.subscribe("t").await, Err(Error::Closed)));
365 }
366
367 #[tokio::test]
368 async fn recv_before_connect_reports_closed() {
369 let mut transport = MqttTransport::new(MqttConfig::new("c", "localhost", 1883));
370 assert!(matches!(transport.recv().await, Err(Error::Closed)));
371 }
372
373 #[tokio::test]
374 async fn connect_to_unavailable_broker_fails() {
375 let config = MqttConfig::new("c", "127.0.0.1", unused_port());
376 let mut transport = MqttTransport::new(config.keep_alive(Duration::from_secs(1)));
377 assert!(matches!(
378 transport.connect().await,
379 Err(Error::Transport(_))
380 ));
381 assert!(!transport.is_connected());
382 }
383}