pamoja_mqtt/lib.rs
1//! MQTT transport for the pamoja SDK.
2//!
3//! [`MqttTransport`] implements the core [`Transport`] and [`Receive`] traits on
4//! top of the pure-Rust [`rumqttc`] client, so an application can publish to and
5//! subscribe from an MQTT broker through the same protocol-agnostic surface it uses
6//! 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`](Receive::recv) drains. Publishing and subscribing use the default
12//! [`QualityOfService`] configured on the transport.
13//!
14//! The client speaks MQTT 3.1.1 with a clean session, so a connection starts with
15//! no subscriptions and a reconnect places them again. Topics and filters are
16//! checked against the rules of the OASIS MQTT 3.1.1 standard, section 4.7, before
17//! anything is sent, and a message too large for the connection's packet limit is
18//! refused rather than ending the connection.
19//!
20//! Beyond [`send`](Transport::send), [`publish`](MqttTransport::publish) takes the
21//! quality of service and retain flag of one message and returns a [`Delivery`] that
22//! settles when the broker acknowledges it. A configuration can carry a username and
23//! password, a last-will message the broker publishes if the client drops off without
24//! saying goodbye, and, with the default `tls` feature, [`Tls`] settings for a broker
25//! on port 8883.
26//!
27//! # Examples
28//!
29//! ```no_run
30//! use pamoja_core::{Receive, Transport};
31//! use pamoja_mqtt::{MqttConfig, MqttTransport};
32//!
33//! # async fn run() -> pamoja_core::Result<()> {
34//! let mut transport = MqttTransport::new(MqttConfig::new("sensor-1", "localhost", 1883));
35//! transport.connect().await?;
36//! transport.subscribe("sensors/+/temperature").await?;
37//! transport.send("sensors/1/temperature", b"21.5").await?;
38//!
39//! if let Some(message) = transport.recv().await? {
40//! println!("{}: {} bytes", message.topic, message.payload.len());
41//! }
42//! # Ok(())
43//! # }
44//! ```
45
46mod delivery;
47#[cfg(feature = "tls")]
48mod tls;
49
50use std::sync::atomic::{AtomicBool, Ordering};
51use std::sync::Arc;
52use std::time::Duration;
53
54pub use delivery::Delivery;
55pub use pamoja_core::Message;
56use pamoja_core::{Error, Receive, Result, Transport};
57use rumqttc::{
58 AsyncClient, ClientError, ConnectReturnCode, ConnectionError, Event, LastWill, MqttOptions,
59 Outgoing, Packet, QoS,
60};
61#[cfg(feature = "tls")]
62pub use tls::Tls;
63use tokio::sync::{mpsc, oneshot};
64use tokio::task::JoinHandle;
65
66use delivery::{InFlight, Requested};
67
68/// The delivery guarantee applied to published and subscribed messages.
69///
70/// These map one-to-one onto the MQTT protocol's quality-of-service levels.
71#[derive(Clone, Copy, Debug, PartialEq, Eq)]
72pub enum QualityOfService {
73 /// Fire and forget: the broker does not acknowledge delivery.
74 AtMostOnce,
75 /// The message is delivered at least once and acknowledged.
76 AtLeastOnce,
77 /// The message is delivered exactly once via a four-step handshake.
78 ExactlyOnce,
79}
80
81impl From<QualityOfService> for QoS {
82 fn from(value: QualityOfService) -> Self {
83 match value {
84 QualityOfService::AtMostOnce => QoS::AtMostOnce,
85 QualityOfService::AtLeastOnce => QoS::AtLeastOnce,
86 QualityOfService::ExactlyOnce => QoS::ExactlyOnce,
87 }
88 }
89}
90
91/// The largest packet a connection sends or accepts unless configured otherwise, in
92/// bytes.
93pub const DEFAULT_MAX_PACKET_SIZE: usize = 10 * 1024;
94
95/// A message the broker publishes on the client's behalf if the client's connection ends
96/// without a `DISCONNECT`: the network dropped, the power failed, or the keep-alive ran
97/// out.
98///
99/// A device that publishes `online` to a status topic when it connects usually leaves a
100/// will of `offline` on the same topic, retained, so anyone watching sees it go (OASIS MQTT
101/// 3.1.1, section 3.1.2.5). A client that disconnects on purpose leaves no will behind.
102///
103/// # Examples
104///
105/// ```
106/// use pamoja_mqtt::{MqttConfig, QualityOfService, Will};
107///
108/// let will = Will::new("sites/pump-3/status", b"offline")
109/// .qos(QualityOfService::AtLeastOnce)
110/// .retained();
111/// let config = MqttConfig::new("pump-3", "localhost", 1883).last_will(will);
112/// assert_eq!(config.will().map(Will::topic), Some("sites/pump-3/status"));
113/// ```
114#[derive(Clone, Debug, PartialEq, Eq)]
115pub struct Will {
116 topic: String,
117 payload: Vec<u8>,
118 qos: QualityOfService,
119 retain: bool,
120}
121
122impl Will {
123 /// Creates a will to publish to a topic, at QoS 0 and not retained.
124 ///
125 /// # Arguments
126 ///
127 /// * `topic` - where the broker publishes it; a topic name, with no wildcard.
128 /// * `payload` - what it publishes.
129 ///
130 /// # Returns
131 ///
132 /// The will.
133 pub fn new(topic: impl Into<String>, payload: impl Into<Vec<u8>>) -> Will {
134 Will {
135 topic: topic.into(),
136 payload: payload.into(),
137 qos: QualityOfService::AtMostOnce,
138 retain: false,
139 }
140 }
141
142 /// Sets the quality of service the broker publishes the will at.
143 ///
144 /// # Arguments
145 ///
146 /// * `qos` - the delivery guarantee.
147 ///
148 /// # Returns
149 ///
150 /// The updated will, for chaining.
151 pub fn qos(mut self, qos: QualityOfService) -> Will {
152 self.qos = qos;
153 self
154 }
155
156 /// Has the broker retain the will, so a client that subscribes later still sees it.
157 ///
158 /// # Returns
159 ///
160 /// The updated will, for chaining.
161 pub fn retained(mut self) -> Will {
162 self.retain = true;
163 self
164 }
165
166 /// Returns the topic the will is published to.
167 ///
168 /// # Returns
169 ///
170 /// The topic.
171 pub fn topic(&self) -> &str {
172 &self.topic
173 }
174
175 /// Returns what the will publishes.
176 ///
177 /// # Returns
178 ///
179 /// The payload.
180 pub fn payload(&self) -> &[u8] {
181 &self.payload
182 }
183
184 /// Returns the quality of service the will is published at.
185 ///
186 /// # Returns
187 ///
188 /// The delivery guarantee.
189 pub fn quality_of_service(&self) -> QualityOfService {
190 self.qos
191 }
192
193 /// Returns whether the broker retains the will.
194 ///
195 /// # Returns
196 ///
197 /// `true` for a retained will.
198 pub fn is_retained(&self) -> bool {
199 self.retain
200 }
201}
202
203/// How one message is published: its quality of service, and whether the broker keeps it.
204///
205/// A retained message is the one the broker hands to every client that subscribes to the
206/// topic later, so a new dashboard sees the last reading at once rather than waiting for the
207/// next (OASIS MQTT 3.1.1, section 3.3.1.3). Publishing an empty retained message clears the
208/// one the broker holds.
209#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
210pub struct PublishOptions {
211 qos: Option<QualityOfService>,
212 retain: bool,
213}
214
215impl PublishOptions {
216 /// Options that publish at the connection's quality of service, not retained.
217 ///
218 /// # Returns
219 ///
220 /// The options.
221 pub fn new() -> PublishOptions {
222 PublishOptions::default()
223 }
224
225 /// Publishes at a quality of service other than the connection's.
226 ///
227 /// # Arguments
228 ///
229 /// * `qos` - the delivery guarantee for this message.
230 ///
231 /// # Returns
232 ///
233 /// The updated options, for chaining.
234 pub fn qos(mut self, qos: QualityOfService) -> PublishOptions {
235 self.qos = Some(qos);
236 self
237 }
238
239 /// Has the broker retain the message for clients that subscribe later.
240 ///
241 /// # Returns
242 ///
243 /// The updated options, for chaining.
244 pub fn retained(mut self) -> PublishOptions {
245 self.retain = true;
246 self
247 }
248
249 /// Returns whether the message is retained.
250 ///
251 /// # Returns
252 ///
253 /// `true` for a retained message.
254 pub fn is_retained(&self) -> bool {
255 self.retain
256 }
257}
258
259/// Connection settings for an [`MqttTransport`].
260///
261/// Construct with [`MqttConfig::new`] and refine with the chained setters; every
262/// field has a sensible default so only the broker address and client id are
263/// required.
264#[derive(Clone)]
265pub struct MqttConfig {
266 client_id: String,
267 host: String,
268 port: u16,
269 keep_alive: Duration,
270 capacity: usize,
271 qos: QualityOfService,
272 max_packet_size: usize,
273 credentials: Option<(String, String)>,
274 will: Option<Will>,
275 #[cfg(feature = "tls")]
276 tls: Option<Tls>,
277}
278
279// A password never reaches a log or a panic message through this type.
280impl core::fmt::Debug for MqttConfig {
281 fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
282 let mut debug = f.debug_struct("MqttConfig");
283 debug
284 .field("client_id", &self.client_id)
285 .field("host", &self.host)
286 .field("port", &self.port)
287 .field("keep_alive", &self.keep_alive)
288 .field("capacity", &self.capacity)
289 .field("qos", &self.qos)
290 .field("max_packet_size", &self.max_packet_size)
291 .field(
292 "username",
293 &self.credentials.as_ref().map(|(username, _)| username),
294 )
295 .field("will", &self.will);
296 #[cfg(feature = "tls")]
297 debug.field("tls", &self.tls);
298 debug.finish_non_exhaustive()
299 }
300}
301
302impl MqttConfig {
303 /// Creates a configuration for the given client id and broker address.
304 ///
305 /// # Arguments
306 ///
307 /// * `client_id` - the MQTT client identifier presented to the broker.
308 /// * `host` - the broker hostname or IP address.
309 /// * `port` - the broker TCP port, conventionally `1883` for plaintext MQTT.
310 ///
311 /// # Returns
312 ///
313 /// A configuration with a 30-second keep-alive, a request capacity of 64, a
314 /// default quality of service of [`QualityOfService::AtLeastOnce`], and a
315 /// packet limit of [`DEFAULT_MAX_PACKET_SIZE`].
316 pub fn new(client_id: impl Into<String>, host: impl Into<String>, port: u16) -> Self {
317 Self {
318 client_id: client_id.into(),
319 host: host.into(),
320 port,
321 keep_alive: Duration::from_secs(30),
322 capacity: 64,
323 qos: QualityOfService::AtLeastOnce,
324 max_packet_size: DEFAULT_MAX_PACKET_SIZE,
325 credentials: None,
326 will: None,
327 #[cfg(feature = "tls")]
328 tls: None,
329 }
330 }
331
332 /// Signs in to the broker with a username and password.
333 ///
334 /// The password travels in the clear unless the connection uses TLS, so a broker
335 /// reached over a network worth protecting wants both.
336 ///
337 /// # Arguments
338 ///
339 /// * `username` - the name the broker knows the client by.
340 /// * `password` - its password.
341 ///
342 /// # Returns
343 ///
344 /// The updated configuration, for chaining.
345 pub fn credentials(mut self, username: impl Into<String>, password: impl Into<String>) -> Self {
346 self.credentials = Some((username.into(), password.into()));
347 self
348 }
349
350 /// Leaves a message for the broker to publish if the connection ends without a
351 /// `DISCONNECT`.
352 ///
353 /// # Arguments
354 ///
355 /// * `will` - the message, its topic, and how it is published.
356 ///
357 /// # Returns
358 ///
359 /// The updated configuration, for chaining.
360 pub fn last_will(mut self, will: Will) -> Self {
361 self.will = Some(will);
362 self
363 }
364
365 /// Secures the connection with TLS.
366 ///
367 /// # Arguments
368 ///
369 /// * `tls` - the authorities to trust, and any client certificate to present.
370 ///
371 /// # Returns
372 ///
373 /// The updated configuration, for chaining.
374 #[cfg(feature = "tls")]
375 pub fn tls(mut self, tls: Tls) -> Self {
376 self.tls = Some(tls);
377 self
378 }
379
380 /// Returns the username the client signs in with.
381 ///
382 /// # Returns
383 ///
384 /// The username, or `None` for a client that does not sign in.
385 pub fn username(&self) -> Option<&str> {
386 self.credentials
387 .as_ref()
388 .map(|(username, _)| username.as_str())
389 }
390
391 /// Returns the will the broker holds for this client.
392 ///
393 /// # Returns
394 ///
395 /// The will, or `None` when there is none.
396 pub fn will(&self) -> Option<&Will> {
397 self.will.as_ref()
398 }
399
400 /// Returns whether the connection uses TLS.
401 ///
402 /// # Returns
403 ///
404 /// `true` once [`tls`](MqttConfig::tls) has been set.
405 pub fn uses_tls(&self) -> bool {
406 #[cfg(feature = "tls")]
407 return self.tls.is_some();
408 #[cfg(not(feature = "tls"))]
409 return false;
410 }
411
412 /// Sets the largest packet the connection sends or accepts, in bytes.
413 ///
414 /// A packet is a message's topic and payload plus a few bytes of framing. A send
415 /// whose packet would be larger is refused and the connection stays up, but a
416 /// larger packet arriving from the broker ends the connection, so every client
417 /// that shares a topic needs a limit that fits it. MQTT itself allows a packet of
418 /// up to 268,435,455 bytes after its fixed header.
419 ///
420 /// # Arguments
421 ///
422 /// * `bytes` - the limit, which applies to both directions.
423 ///
424 /// # Returns
425 ///
426 /// The updated configuration, for chaining.
427 pub fn max_packet_size(mut self, bytes: usize) -> Self {
428 self.max_packet_size = bytes;
429 self
430 }
431
432 /// Sets the keep-alive interval used to hold the connection open.
433 ///
434 /// # Arguments
435 ///
436 /// * `interval` - how often the client pings the broker when otherwise idle.
437 ///
438 /// # Returns
439 ///
440 /// The updated configuration, for chaining.
441 pub fn keep_alive(mut self, interval: Duration) -> Self {
442 self.keep_alive = interval;
443 self
444 }
445
446 /// Sets the bound on outstanding client requests buffered toward the broker.
447 ///
448 /// # Arguments
449 ///
450 /// * `capacity` - the request channel capacity; values below one are clamped
451 /// to one.
452 ///
453 /// # Returns
454 ///
455 /// The updated configuration, for chaining.
456 pub fn capacity(mut self, capacity: usize) -> Self {
457 self.capacity = capacity.max(1);
458 self
459 }
460
461 /// Sets the default quality of service for publishes and subscriptions.
462 ///
463 /// # Arguments
464 ///
465 /// * `qos` - the delivery guarantee applied by [`send`](Transport::send) and
466 /// [`subscribe`](Transport::subscribe).
467 ///
468 /// # Returns
469 ///
470 /// The updated configuration, for chaining.
471 pub fn qos(mut self, qos: QualityOfService) -> Self {
472 self.qos = qos;
473 self
474 }
475}
476
477/// The messages arriving on a transport's connection, readable apart from the transport.
478///
479/// [`recv`](Receive::recv) borrows the transport for as long as it waits, so a program that
480/// waits for commands in one task while it publishes readings from another takes an inbox
481/// with [`MqttTransport::inbox`] and waits on that instead. The inbox follows the transport
482/// across reconnects.
483#[derive(Clone, Default)]
484pub struct Inbox(Arc<tokio::sync::Mutex<Option<mpsc::UnboundedReceiver<Result<Message>>>>>);
485
486impl Inbox {
487 /// Awaits the next message from any subscribed topic.
488 ///
489 /// # Returns
490 ///
491 /// `Some(message)` for the next queued message, or `None` once the connection has
492 /// ended and its reason has been reported.
493 ///
494 /// # Errors
495 ///
496 /// Returns [`Error::Closed`] if the transport is not connected, or
497 /// [`Error::Transport`] once, saying why, when the connection ended on its own.
498 pub async fn recv(&self) -> Result<Option<Message>> {
499 let mut held = self.0.lock().await;
500 let incoming = held.as_mut().ok_or(Error::Closed)?;
501 incoming.recv().await.transpose()
502 }
503
504 async fn replace(&self, incoming: Option<mpsc::UnboundedReceiver<Result<Message>>>) {
505 *self.0.lock().await = incoming;
506 }
507}
508
509impl core::fmt::Debug for Inbox {
510 fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
511 f.debug_struct("Inbox").finish_non_exhaustive()
512 }
513}
514
515/// An MQTT client that implements the core [`Transport`] and [`Receive`] traits.
516///
517/// A transport is created disconnected; [`connect`](Transport::connect) opens the
518/// link and spawns the background task that runs the MQTT event loop for the life
519/// of the connection. Inbound messages are queued and read with
520/// [`recv`](Receive::recv).
521///
522/// A connection can end without being asked to: the broker restarts, another client
523/// connects with the same client id, or a packet over the limit arrives. From then
524/// on [`is_connected`](MqttTransport::is_connected) is `false`, a send answers
525/// [`Error::Closed`], and the next receive reports why the connection ended, until
526/// [`connect`](Transport::connect) opens a new one.
527pub struct MqttTransport {
528 config: MqttConfig,
529 client: Option<AsyncClient>,
530 inbox: Inbox,
531 pump: Option<JoinHandle<()>>,
532 ended: Arc<AtomicBool>,
533 requested: Requested,
534}
535
536impl MqttTransport {
537 /// Creates a transport from the given configuration without connecting.
538 ///
539 /// # Arguments
540 ///
541 /// * `config` - the broker connection settings.
542 ///
543 /// # Returns
544 ///
545 /// A disconnected transport ready for [`connect`](Transport::connect).
546 pub fn new(config: MqttConfig) -> Self {
547 Self {
548 config,
549 client: None,
550 inbox: Inbox::default(),
551 pump: None,
552 ended: Arc::new(AtomicBool::new(false)),
553 requested: Requested::default(),
554 }
555 }
556
557 /// Publishes a message with options of its own, and follows it to the broker.
558 ///
559 /// The call returns once the message is queued for the broker, as
560 /// [`send`](Transport::send) does. The [`Delivery`] it hands back settles when the
561 /// broker acknowledges the message, so a caller that needs to know the message arrived
562 /// awaits it, and one that does not drops it.
563 ///
564 /// # Arguments
565 ///
566 /// * `topic` - the topic to publish to, with no wildcard.
567 /// * `payload` - the message.
568 /// * `options` - the quality of service and retain flag for this message.
569 ///
570 /// # Returns
571 ///
572 /// The delivery to await for the broker's acknowledgment.
573 ///
574 /// # Errors
575 ///
576 /// Returns [`Error::Transport`] if MQTT does not allow the topic or the message is over
577 /// the connection's packet limit, both before anything is sent, or [`Error::Closed`] if
578 /// the transport is not connected.
579 ///
580 /// # Examples
581 ///
582 /// ```no_run
583 /// use pamoja_core::Transport;
584 /// use pamoja_mqtt::{MqttConfig, MqttTransport, PublishOptions};
585 ///
586 /// # async fn run() -> pamoja_core::Result<()> {
587 /// let mut transport = MqttTransport::new(MqttConfig::new("sensor-1", "localhost", 1883));
588 /// transport.connect().await?;
589 ///
590 /// // The last reading stays with the broker for any dashboard that subscribes later,
591 /// // and the call waits until the broker says it has it.
592 /// transport
593 /// .publish("sites/tank-2/level", b"73", PublishOptions::new().retained())
594 /// .await?
595 /// .confirmed()
596 /// .await?;
597 /// # Ok(())
598 /// # }
599 /// ```
600 pub async fn publish(
601 &mut self,
602 topic: &str,
603 payload: &[u8],
604 options: PublishOptions,
605 ) -> Result<Delivery> {
606 let client = self.live()?.clone();
607 check_topic(topic)?;
608 let qos = options.qos.unwrap_or(self.config.qos);
609 let size = publish_size(topic, payload.len(), qos);
610 if size > self.config.max_packet_size {
611 return Err(Error::Transport(format!(
612 "the message makes a {size}-byte packet, over this connection's {}-byte limit",
613 self.config.max_packet_size
614 )));
615 }
616 let delivery = self.requested.push(qos.into());
617 if let Err(error) = client
618 .publish(topic, qos.into(), options.retain, payload.to_vec())
619 .await
620 {
621 self.requested.withdraw();
622 return Err(map_client_error(error));
623 }
624 Ok(delivery)
625 }
626
627 /// Returns a handle on the messages arriving on this transport's connection.
628 ///
629 /// # Returns
630 ///
631 /// The inbox, which a task can wait on while another publishes through the transport.
632 pub fn inbox(&self) -> Inbox {
633 self.inbox.clone()
634 }
635
636 /// Reports whether the transport currently holds an active connection.
637 ///
638 /// # Returns
639 ///
640 /// `true` once [`connect`](Transport::connect) has succeeded, until
641 /// [`disconnect`](MqttTransport::disconnect) is called or the connection ends
642 /// on its own.
643 pub fn is_connected(&self) -> bool {
644 self.live().is_ok()
645 }
646
647 /// The client, while its connection is up.
648 ///
649 /// The event loop marks the connection ended before it reports why, so a
650 /// receive that has seen the report never finds the transport still up.
651 fn live(&self) -> Result<&AsyncClient> {
652 match (&self.client, &self.pump) {
653 (Some(client), Some(pump))
654 if !pump.is_finished() && !self.ended.load(Ordering::SeqCst) =>
655 {
656 Ok(client)
657 }
658 _ => Err(Error::Closed),
659 }
660 }
661
662 /// Closes the connection and stops the background event loop.
663 ///
664 /// The client says goodbye with a `DISCONNECT`, so the broker discards the will
665 /// rather than publishing it (OASIS MQTT 3.1.1, section 3.14). Calling this on a
666 /// transport that is not connected is a no-op. A transport that is dropped without
667 /// disconnecting closes its connection without a goodbye, and the broker publishes
668 /// its will.
669 ///
670 /// # Returns
671 ///
672 /// `Ok(())` once the goodbye has gone out, or a second has passed without the event
673 /// loop sending it, and the loop has been stopped.
674 ///
675 /// # Errors
676 ///
677 /// This call is best-effort and does not surface broker errors raised while
678 /// tearing down, so it currently always returns `Ok(())`.
679 pub async fn disconnect(&mut self) -> Result<()> {
680 let pump = self.pump.take();
681 if let Some(client) = self.client.take() {
682 if client.disconnect().await.is_ok() {
683 if let Some(pump) = &pump {
684 let deadline = tokio::time::Instant::now() + Duration::from_secs(1);
685 while !pump.is_finished() && tokio::time::Instant::now() < deadline {
686 tokio::time::sleep(Duration::from_millis(10)).await;
687 }
688 }
689 }
690 }
691 if let Some(pump) = pump {
692 pump.abort();
693 }
694 self.inbox.replace(None).await;
695 self.requested = Requested::default();
696 Ok(())
697 }
698}
699
700impl Drop for MqttTransport {
701 fn drop(&mut self) {
702 if let Some(pump) = self.pump.take() {
703 pump.abort();
704 }
705 }
706}
707
708impl Transport for MqttTransport {
709 /// Opens a connection to the broker, closing any the transport already holds.
710 ///
711 /// The session is clean, so subscriptions from an earlier connection are gone
712 /// and have to be placed again.
713 ///
714 /// # Returns
715 ///
716 /// `Ok(())` once the broker has accepted the connection.
717 ///
718 /// # Errors
719 ///
720 /// Returns [`Error::Transport`] if the broker cannot be reached or refuses the
721 /// connection.
722 async fn connect(&mut self) -> Result<()> {
723 self.disconnect().await?;
724 let mut options = MqttOptions::new(
725 self.config.client_id.clone(),
726 self.config.host.clone(),
727 self.config.port,
728 );
729 options.set_keep_alive(self.config.keep_alive);
730 options.set_max_packet_size(self.config.max_packet_size, self.config.max_packet_size);
731 if let Some((username, password)) = &self.config.credentials {
732 options.set_credentials(username.clone(), password.clone());
733 }
734 if let Some(will) = &self.config.will {
735 check_topic(&will.topic)?;
736 options.set_last_will(LastWill::new(
737 will.topic.clone(),
738 will.payload.clone(),
739 will.qos.into(),
740 will.retain,
741 ));
742 }
743 #[cfg(feature = "tls")]
744 if let Some(tls) = &self.config.tls {
745 options.set_transport(rumqttc::Transport::tls_with_config(
746 rumqttc::TlsConfiguration::Rustls(tls.client_config()?),
747 ));
748 }
749
750 let (client, mut eventloop) = AsyncClient::new(options, self.config.capacity);
751 let (tx, rx) = mpsc::unbounded_channel();
752 let (ready_tx, ready_rx) = oneshot::channel::<Result<()>>();
753 let ended = Arc::new(AtomicBool::new(false));
754 let marked = Arc::clone(&ended);
755 let requested = Requested::default();
756 let reported = requested.clone();
757
758 let pump = tokio::spawn(async move {
759 let mut ready_tx = Some(ready_tx);
760 let mut in_flight = InFlight::default();
761 loop {
762 match eventloop.poll().await {
763 Ok(Event::Incoming(Packet::ConnAck(_))) => {
764 if let Some(ready_tx) = ready_tx.take() {
765 let _ = ready_tx.send(Ok(()));
766 }
767 }
768 Ok(Event::Incoming(Packet::Publish(publish))) => {
769 let message = Message::new(publish.topic, publish.payload.to_vec());
770 if tx.send(Ok(message)).is_err() {
771 break;
772 }
773 }
774 Ok(Event::Outgoing(Outgoing::Publish(pkid))) => {
775 in_flight.sent(&reported, pkid);
776 }
777 Ok(Event::Incoming(Packet::PubAck(ack))) => in_flight.acknowledged(ack.pkid),
778 Ok(Event::Incoming(Packet::PubComp(comp))) => {
779 in_flight.acknowledged(comp.pkid);
780 }
781 Ok(Event::Outgoing(Outgoing::Disconnect)) => {
782 marked.store(true, Ordering::SeqCst);
783 in_flight.fail(&reported, "the client disconnected");
784 break;
785 }
786 Ok(_) => {}
787 Err(err) => {
788 marked.store(true, Ordering::SeqCst);
789 in_flight.fail(&reported, &err.to_string());
790 match ready_tx.take() {
791 Some(ready_tx) => {
792 let _ = ready_tx.send(Err(map_connection_error(err)));
793 }
794 None => {
795 let _ = tx.send(Err(Error::Transport(format!(
796 "the connection to the broker ended: {err}"
797 ))));
798 }
799 }
800 break;
801 }
802 }
803 }
804 });
805
806 match ready_rx.await {
807 Ok(Ok(())) => {
808 self.client = Some(client);
809 self.inbox.replace(Some(rx)).await;
810 self.pump = Some(pump);
811 self.ended = ended;
812 self.requested = requested;
813 Ok(())
814 }
815 Ok(Err(err)) => {
816 pump.abort();
817 Err(err)
818 }
819 Err(_) => {
820 pump.abort();
821 Err(Error::Transport(
822 "event loop closed before the connection was established".into(),
823 ))
824 }
825 }
826 }
827
828 /// Publishes a payload to a topic under the configured quality of service.
829 ///
830 /// # Returns
831 ///
832 /// `Ok(())` once the message is queued for the broker, which is before the broker
833 /// acknowledges it; [`publish`](MqttTransport::publish) waits for that.
834 ///
835 /// # Errors
836 ///
837 /// Returns [`Error::Transport`] if MQTT does not allow the topic or the message
838 /// is over the connection's packet limit, both before anything is sent, or
839 /// [`Error::Closed`] if the transport is not connected.
840 async fn send(&mut self, topic: &str, payload: &[u8]) -> Result<()> {
841 self.publish(topic, payload, PublishOptions::new())
842 .await
843 .map(drop)
844 }
845
846 /// Subscribes to a topic filter under the configured quality of service.
847 ///
848 /// # Returns
849 ///
850 /// `Ok(())` once the subscription is queued for the broker.
851 ///
852 /// # Errors
853 ///
854 /// Returns [`Error::Transport`] if the filter places a wildcard where MQTT does
855 /// not allow one, or [`Error::Closed`] if the transport is not connected.
856 async fn subscribe(&mut self, topic: &str) -> Result<()> {
857 let client = self.live()?;
858 check_filter(topic)?;
859 client
860 .subscribe(topic, self.config.qos.into())
861 .await
862 .map_err(map_client_error)
863 }
864}
865
866impl Receive for MqttTransport {
867 /// Awaits the next message from any subscribed topic.
868 ///
869 /// # Returns
870 ///
871 /// `Some(message)` for the next queued message, or `None` once the connection
872 /// has ended and its reason has been reported.
873 ///
874 /// # Errors
875 ///
876 /// Returns [`Error::Closed`] if the transport is not connected, or
877 /// [`Error::Transport`] once, saying why, when the connection ended on its own.
878 async fn recv(&mut self) -> Result<Option<Message>> {
879 self.inbox.recv().await
880 }
881}
882
883/// Refuses a topic MQTT does not allow a message to be published to.
884///
885/// The rules are the OASIS MQTT 3.1.1 standard's: a topic name holds no wildcard
886/// (section 4.7.1) and is between one and 65,535 bytes of UTF-8 with no null
887/// character (section 4.7.3).
888fn check_topic(topic: &str) -> Result<()> {
889 if topic.contains(['+', '#']) {
890 return Err(Error::Transport(format!(
891 "the topic {topic} holds a wildcard, which only a subscription may use"
892 )));
893 }
894 check_length(topic)
895}
896
897/// Refuses a filter that places a wildcard where MQTT does not allow one.
898///
899/// The rules are the OASIS MQTT 3.1.1 standard's, section 4.7.1: `#` stands alone
900/// in the last level, and `+` fills a whole level.
901fn check_filter(filter: &str) -> Result<()> {
902 let levels: Vec<&str> = filter.split('/').collect();
903 let last = levels.len() - 1;
904 let misplaced = levels.iter().enumerate().any(|(at, level)| {
905 let hash = level.contains('#') && (*level != "#" || at != last);
906 let plus = level.contains('+') && *level != "+";
907 hash || plus
908 });
909 if misplaced {
910 return Err(Error::Transport(format!(
911 "the filter {filter} places a wildcard where MQTT does not allow one: \
912 # only alone in the last level, + only as a whole level"
913 )));
914 }
915 check_length(filter)
916}
917
918/// Refuses a topic or filter outside the lengths and characters MQTT allows.
919fn check_length(topic: &str) -> Result<()> {
920 if topic.is_empty() {
921 return Err(Error::Transport(
922 "an MQTT topic has at least one character".into(),
923 ));
924 }
925 if topic.contains('\0') {
926 return Err(Error::Transport(
927 "an MQTT topic cannot hold a null character".into(),
928 ));
929 }
930 if topic.len() > usize::from(u16::MAX) {
931 return Err(Error::Transport(
932 "an MQTT topic is at most 65,535 bytes".into(),
933 ));
934 }
935 Ok(())
936}
937
938/// The size of the packet that publishes a message, framing included.
939///
940/// A publish is a fixed-header byte, a remaining length of one to four bytes, a
941/// two-byte topic length, the topic, a two-byte packet identifier above QoS 0, and
942/// the payload (OASIS MQTT 3.1.1, sections 2.2.3 and 3.3). This is the size the
943/// event loop checks against the limit, so a send that would fail there is refused
944/// before it can end the connection.
945fn publish_size(topic: &str, payload: usize, qos: QualityOfService) -> usize {
946 let packet_id = if qos == QualityOfService::AtMostOnce {
947 0
948 } else {
949 2
950 };
951 let remaining = 2 + topic.len() + packet_id + payload;
952 let length_bytes = match remaining {
953 0..=127 => 1,
954 128..=16_383 => 2,
955 16_384..=2_097_151 => 3,
956 _ => 4,
957 };
958 1 + length_bytes + remaining
959}
960
961/// Maps a `rumqttc` client error onto the shared transport error.
962///
963/// Topics and filters are checked before a request is made, so the one failure
964/// left is a request the event loop can no longer take: the connection has ended.
965fn map_client_error(_: ClientError) -> Error {
966 Error::Closed
967}
968
969/// Maps a `rumqttc` event-loop error onto the shared transport error.
970///
971/// A broker that turns the client away says why in its `CONNACK` return code (OASIS MQTT
972/// 3.1.1, section 3.2.2.3), and the two about who the client is are authentication errors.
973fn map_connection_error(err: ConnectionError) -> Error {
974 match err {
975 ConnectionError::ConnectionRefused(ConnectReturnCode::BadUserNamePassword) => {
976 Error::Auth("the broker refused the username or password".into())
977 }
978 ConnectionError::ConnectionRefused(ConnectReturnCode::NotAuthorized) => {
979 Error::Auth("the broker does not authorize this client".into())
980 }
981 ConnectionError::ConnectionRefused(ConnectReturnCode::BadClientId) => {
982 Error::Transport("the broker refused the client id".into())
983 }
984 other => Error::Transport(other.to_string()),
985 }
986}
987
988#[cfg(test)]
989mod tests {
990 use super::*;
991 use std::net::TcpListener;
992
993 /// Returns a TCP port with no listener bound, for negative connection tests.
994 fn unused_port() -> u16 {
995 let listener = TcpListener::bind("127.0.0.1:0").expect("bind ephemeral port");
996 listener.local_addr().expect("local addr").port()
997 }
998
999 #[test]
1000 fn quality_of_service_maps_to_rumqttc() {
1001 assert_eq!(QoS::from(QualityOfService::AtMostOnce), QoS::AtMostOnce);
1002 assert_eq!(QoS::from(QualityOfService::AtLeastOnce), QoS::AtLeastOnce);
1003 assert_eq!(QoS::from(QualityOfService::ExactlyOnce), QoS::ExactlyOnce);
1004 }
1005
1006 #[test]
1007 fn capacity_is_clamped_to_at_least_one() {
1008 let config = MqttConfig::new("c", "localhost", 1883).capacity(0);
1009 assert_eq!(config.capacity, 1);
1010 }
1011
1012 #[test]
1013 fn filters_follow_the_examples_in_section_4_7_1() {
1014 for valid in [
1015 "sport/tennis/player1/#",
1016 "#",
1017 "sport/tennis/#",
1018 "+",
1019 "+/tennis/#",
1020 "sport/+/player1",
1021 "/+",
1022 "+/+",
1023 ] {
1024 assert!(check_filter(valid).is_ok(), "{valid} is a valid filter");
1025 }
1026 for invalid in ["sport/tennis#", "sport/tennis/#/ranking", "sport+"] {
1027 assert!(
1028 matches!(check_filter(invalid), Err(Error::Transport(_))),
1029 "{invalid} is not a valid filter"
1030 );
1031 }
1032 }
1033
1034 #[test]
1035 fn a_topic_to_publish_to_holds_no_wildcard_and_has_a_length() {
1036 assert!(check_topic("sport/tennis/player1").is_ok());
1037 assert!(check_topic("/finance").is_ok());
1038 for refused in ["sport/+/player1", "sport/#", "", "a\0b"] {
1039 assert!(
1040 matches!(check_topic(refused), Err(Error::Transport(_))),
1041 "{refused:?} is refused"
1042 );
1043 }
1044 let longest = "t".repeat(65_535);
1045 assert!(check_topic(&longest).is_ok());
1046 assert!(check_topic(&format!("{longest}t")).is_err());
1047 }
1048
1049 #[test]
1050 fn a_publish_packet_grows_its_length_field_at_the_table_2_4_boundaries() {
1051 use QualityOfService::{AtLeastOnce, AtMostOnce};
1052 assert_eq!(publish_size("a", 124, AtMostOnce), 1 + 1 + 127);
1053 assert_eq!(publish_size("a", 125, AtMostOnce), 1 + 2 + 128);
1054 assert_eq!(publish_size("a", 122, AtLeastOnce), 1 + 1 + 127);
1055 assert_eq!(publish_size("a", 16_380, AtMostOnce), 1 + 2 + 16_383);
1056 assert_eq!(publish_size("a", 16_381, AtMostOnce), 1 + 3 + 16_384);
1057 assert_eq!(publish_size("a", 2_097_148, AtMostOnce), 1 + 3 + 2_097_151);
1058 assert_eq!(publish_size("a", 2_097_149, AtMostOnce), 1 + 4 + 2_097_152);
1059 }
1060
1061 #[test]
1062 fn a_temperature_reading_packs_into_the_sizes_the_guide_lists() {
1063 use QualityOfService::{AtLeastOnce, AtMostOnce};
1064 let topic = "sensors/1/temperature";
1065 assert_eq!(publish_size(topic, 100, AtMostOnce), 125);
1066 assert_eq!(publish_size(topic, 100, AtLeastOnce), 127);
1067 assert_eq!(publish_size(topic, 10_000, AtMostOnce), 10_026);
1068 assert_eq!(publish_size(topic, 10_000, AtLeastOnce), 10_028);
1069 assert_eq!(publish_size(topic, 10_215, AtMostOnce), 10_241);
1070 assert_eq!(publish_size(topic, 10_215, AtLeastOnce), 10_243);
1071 assert!(publish_size(topic, 10_214, AtMostOnce) <= DEFAULT_MAX_PACKET_SIZE);
1072 }
1073
1074 #[test]
1075 fn the_packet_limit_defaults_to_ten_kibibytes_and_can_be_raised() {
1076 let config = MqttConfig::new("c", "localhost", 1883);
1077 assert_eq!(config.max_packet_size, DEFAULT_MAX_PACKET_SIZE);
1078 assert_eq!(config.max_packet_size(65_536).max_packet_size, 65_536);
1079 }
1080
1081 #[test]
1082 fn a_refusal_about_who_the_client_is_is_an_authentication_error() {
1083 for (code, kind) in [
1084 (
1085 ConnectReturnCode::BadUserNamePassword,
1086 "username or password",
1087 ),
1088 (ConnectReturnCode::NotAuthorized, "does not authorize"),
1089 ] {
1090 match map_connection_error(ConnectionError::ConnectionRefused(code)) {
1091 Error::Auth(reason) => assert!(reason.contains(kind), "{reason}"),
1092 other => panic!("{code:?} is an authentication error, got {other:?}"),
1093 }
1094 }
1095 assert!(matches!(
1096 map_connection_error(ConnectionError::ConnectionRefused(
1097 ConnectReturnCode::BadClientId
1098 )),
1099 Error::Transport(_)
1100 ));
1101 }
1102
1103 #[test]
1104 fn a_will_and_options_carry_what_they_were_given() {
1105 let will = Will::new("sites/pump-3/status", b"offline".to_vec())
1106 .qos(QualityOfService::ExactlyOnce)
1107 .retained();
1108 assert_eq!(will.topic(), "sites/pump-3/status");
1109 assert_eq!(will.payload(), b"offline");
1110 assert_eq!(will.quality_of_service(), QualityOfService::ExactlyOnce);
1111 assert!(will.is_retained());
1112 assert!(!Will::new("t", b"x".to_vec()).is_retained());
1113 assert!(PublishOptions::new().retained().is_retained());
1114 assert!(!PublishOptions::new().is_retained());
1115 }
1116
1117 #[tokio::test]
1118 async fn send_before_connect_reports_closed() {
1119 let mut transport = MqttTransport::new(MqttConfig::new("c", "localhost", 1883));
1120 assert!(matches!(
1121 transport.send("t", b"x").await,
1122 Err(Error::Closed)
1123 ));
1124 }
1125
1126 #[tokio::test]
1127 async fn subscribe_before_connect_reports_closed() {
1128 let mut transport = MqttTransport::new(MqttConfig::new("c", "localhost", 1883));
1129 assert!(matches!(transport.subscribe("t").await, Err(Error::Closed)));
1130 }
1131
1132 #[tokio::test]
1133 async fn recv_before_connect_reports_closed() {
1134 let mut transport = MqttTransport::new(MqttConfig::new("c", "localhost", 1883));
1135 assert!(matches!(transport.recv().await, Err(Error::Closed)));
1136 }
1137
1138 #[tokio::test]
1139 async fn connect_to_unavailable_broker_fails() {
1140 let config = MqttConfig::new("c", "127.0.0.1", unused_port());
1141 let mut transport = MqttTransport::new(config.keep_alive(Duration::from_secs(1)));
1142 assert!(matches!(
1143 transport.connect().await,
1144 Err(Error::Transport(_))
1145 ));
1146 assert!(!transport.is_connected());
1147 }
1148}