mqtt_client_blocking/
lib.rs

1//! A small **blocking** MQTT client on top of [`stem-mqtt-proto`](https://docs.rs/stem-mqtt-proto).
2//!
3//! No async runtime, no TLS stack of its own: it drives the codec over any [`Transport`] (a
4//! `std::net::TcpStream` on a PC or on an ESP32 with ESP-IDF `std`, a TLS stream, a test double).
5//! The caller owns the loop: call [`Client::poll`] regularly, it receives messages and keeps the
6//! connection alive (PINGREQ) by itself.
7//!
8//! ```no_run
9//! use std::net::TcpStream;
10//! use std::time::Duration;
11//! use mqtt_client_blocking::{Client, Options, QoS};
12//!
13//! let stream = TcpStream::connect("192.168.1.10:1883")?;
14//! let mut client = Client::connect(stream, Options::new("weather-c6"))?;
15//! client.subscribe("home/+/command", QoS::AtLeastOnce)?;
16//! client.publish("home/c6/status", b"online", QoS::AtLeastOnce, true)?;
17//! loop {
18//!     if let Some(msg) = client.poll(Duration::from_millis(500))? {
19//!         println!("{}: {}", msg.topic, String::from_utf8_lossy(&msg.payload));
20//!     }
21//! }
22//! # Ok::<(), mqtt_client_blocking::Error>(())
23//! ```
24//!
25//! Scope: QoS 0 and 1 for publishing, QoS 0, 1 and 2 for receiving (a QoS 2 message is delivered
26//! once, when the PUBLISH arrives, and completed with PUBREC/PUBCOMP). Publishing at QoS 2 is
27//! refused with [`Error::Unsupported`]. Reconnecting is the caller's job: after an error, drop the
28//! client and connect again.
29
30#[cfg(doctest)]
31#[doc = include_str!("../README.md")]
32struct ReadmeDoctests;
33
34use std::collections::VecDeque;
35use std::io::{self, Read, Write};
36use std::net::TcpStream;
37use std::time::{Duration, Instant};
38
39use bytes::{Bytes, BytesMut};
40use mqtt_proto::MqttError;
41use mqtt_proto::ack::SimpleAck;
42use mqtt_proto::connect::{ConnectPacket, Will};
43use mqtt_proto::packet::{DisconnectPacket, Packet};
44use mqtt_proto::properties::Properties;
45use mqtt_proto::publish::PublishPacket;
46use mqtt_proto::subscribe::{SubscribeFilter, SubscribePacket, UnsubscribePacket};
47pub use mqtt_proto::{MqttVersion, QoS};
48
49/// How long [`Client::connect`], [`Client::publish`] at QoS 1 and [`Client::subscribe`] wait for
50/// the broker's answer.
51pub const DEFAULT_ACK_TIMEOUT: Duration = Duration::from_secs(5);
52
53/// A byte stream with a read timeout, so that [`Client::poll`] can wait a bounded time.
54pub trait Transport: Read + Write {
55    /// Make reads fail with `WouldBlock` / `TimedOut` after `timeout` (None: block).
56    fn set_read_timeout(&mut self, timeout: Option<Duration>) -> io::Result<()>;
57}
58
59impl Transport for TcpStream {
60    fn set_read_timeout(&mut self, timeout: Option<Duration>) -> io::Result<()> {
61        TcpStream::set_read_timeout(self, timeout)
62    }
63}
64
65/// A message the client sends if the connection is lost without a DISCONNECT.
66#[derive(Debug, Clone, PartialEq, Eq)]
67pub struct WillMessage {
68    pub topic: String,
69    pub payload: Vec<u8>,
70    pub qos: QoS,
71    pub retain: bool,
72}
73
74/// Connection settings.
75#[derive(Debug, Clone)]
76pub struct Options {
77    pub client_id: String,
78    pub username: Option<String>,
79    pub password: Option<Vec<u8>>,
80    /// Protocol version (default 3.1.1).
81    pub version: MqttVersion,
82    /// Keep-alive announced to the broker, in whole seconds (0 turns it off). Default 30 s.
83    pub keep_alive: Duration,
84    /// Start with a clean session (default true).
85    pub clean_session: bool,
86    pub will: Option<WillMessage>,
87    /// How often to send PINGREQ when idle (default: half the keep-alive).
88    pub ping_interval: Option<Duration>,
89    /// How long to wait for CONNACK, PUBACK and SUBACK (default [`DEFAULT_ACK_TIMEOUT`]).
90    pub ack_timeout: Duration,
91}
92
93impl Options {
94    pub fn new(client_id: impl Into<String>) -> Self {
95        Self {
96            client_id: client_id.into(),
97            username: None,
98            password: None,
99            version: MqttVersion::V311,
100            keep_alive: Duration::from_secs(30),
101            clean_session: true,
102            will: None,
103            ping_interval: None,
104            ack_timeout: DEFAULT_ACK_TIMEOUT,
105        }
106    }
107
108    pub fn credentials(
109        mut self,
110        username: impl Into<String>,
111        password: impl Into<Vec<u8>>,
112    ) -> Self {
113        self.username = Some(username.into());
114        self.password = Some(password.into());
115        self
116    }
117
118    pub fn will(mut self, will: WillMessage) -> Self {
119        self.will = Some(will);
120        self
121    }
122}
123
124/// A message received from the broker.
125#[derive(Debug, Clone, PartialEq, Eq)]
126pub struct Message {
127    pub topic: String,
128    pub payload: Vec<u8>,
129    pub qos: QoS,
130    pub retain: bool,
131}
132
133/// What went wrong.
134#[derive(Debug, thiserror::Error)]
135pub enum Error {
136    #[error("i/o error: {0}")]
137    Io(#[from] io::Error),
138    #[error("protocol error: {0}")]
139    Protocol(#[from] MqttError),
140    /// The broker refused the connection (the CONNACK reason code).
141    #[error("connection refused by the broker (code {0})")]
142    Refused(u8),
143    /// The connection was closed by the broker (or the DISCONNECT packet came).
144    #[error("connection closed")]
145    Closed,
146    /// No answer in time (what was awaited).
147    #[error("timed out waiting for {0}")]
148    Timeout(&'static str),
149    /// The broker answered with something that does not fit (what happened).
150    #[error("unexpected answer: {0}")]
151    Unexpected(&'static str),
152    /// A feature this client does not have (publishing at QoS 2).
153    #[error("not supported: {0}")]
154    Unsupported(&'static str),
155    /// The broker rejected a subscription.
156    #[error("subscription refused (code {0})")]
157    SubscribeRefused(u8),
158}
159
160pub type Result<T> = std::result::Result<T, Error>;
161
162/// Traffic counters (for a status screen).
163#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
164pub struct Stats {
165    pub packets_tx: u64,
166    pub packets_rx: u64,
167    pub bytes_tx: u64,
168    pub bytes_rx: u64,
169    /// PUBLISH packets sent / received (application messages).
170    pub messages_tx: u64,
171    pub messages_rx: u64,
172}
173
174/// A connected client.
175#[derive(Debug)]
176pub struct Client<T: Transport> {
177    io: T,
178    version: MqttVersion,
179    buf: BytesMut,
180    keep_alive: Duration,
181    ping_every: Duration,
182    ack_timeout: Duration,
183    last_tx: Instant,
184    ping_sent: Option<Instant>,
185    next_id: u16,
186    stats: Stats,
187    inbox: VecDeque<Message>,
188    session_present: bool,
189}
190
191impl<T: Transport> Client<T> {
192    /// Send CONNECT over `io` and wait for the broker's answer.
193    pub fn connect(mut io: T, opts: Options) -> Result<Self> {
194        let keep_alive_secs = u16::try_from(opts.keep_alive.as_secs()).unwrap_or(u16::MAX);
195        let keep_alive = Duration::from_secs(u64::from(keep_alive_secs));
196        let ping_every = opts.ping_interval.unwrap_or(keep_alive / 2);
197        let connect = ConnectPacket {
198            version: opts.version,
199            client_id: opts.client_id,
200            clean_start: opts.clean_session,
201            keep_alive: keep_alive_secs,
202            username: opts.username,
203            password: opts.password.map(Bytes::from),
204            will: opts.will.map(|w| Will {
205                topic: w.topic,
206                payload: Bytes::from(w.payload),
207                qos: w.qos,
208                retain: w.retain,
209                properties: Properties::default(),
210                delay_interval: 0,
211            }),
212            properties: Properties::default(),
213        };
214        let mut client = Self {
215            io: {
216                io.set_read_timeout(Some(Duration::from_millis(50)))?;
217                io
218            },
219            version: opts.version,
220            buf: BytesMut::new(),
221            keep_alive,
222            ping_every,
223            ack_timeout: opts.ack_timeout,
224            last_tx: Instant::now(),
225            ping_sent: None,
226            next_id: 0,
227            stats: Stats::default(),
228            inbox: VecDeque::new(),
229            session_present: false,
230        };
231        client.send(&Packet::Connect(connect))?;
232        let deadline = Instant::now() + client.ack_timeout;
233        loop {
234            for packet in client.pump(Duration::from_millis(50))? {
235                if let Packet::ConnAck(ack) = packet {
236                    if !ack.reason_code.is_success() {
237                        return Err(Error::Refused(ack.reason_code.0));
238                    }
239                    client.session_present = ack.session_present;
240                    return Ok(client);
241                }
242            }
243            if Instant::now() >= deadline {
244                return Err(Error::Timeout("CONNACK"));
245            }
246        }
247    }
248
249    /// Whether the broker kept a session for this client id.
250    pub fn session_present(&self) -> bool {
251        self.session_present
252    }
253
254    pub fn stats(&self) -> Stats {
255        self.stats
256    }
257
258    /// Publish a message. QoS 1 waits for the broker's PUBACK; QoS 2 is not supported.
259    pub fn publish(&mut self, topic: &str, payload: &[u8], qos: QoS, retain: bool) -> Result<()> {
260        let packet_id = match qos {
261            QoS::AtMostOnce => None,
262            QoS::AtLeastOnce => Some(self.next_packet_id()),
263            QoS::ExactlyOnce => return Err(Error::Unsupported("publishing at QoS 2")),
264        };
265        self.send(&Packet::Publish(PublishPacket {
266            dup: false,
267            qos,
268            retain,
269            topic: topic.to_string(),
270            packet_id,
271            payload: Bytes::copy_from_slice(payload),
272            properties: Properties::default(),
273        }))?;
274        self.stats.messages_tx += 1;
275        if let Some(id) = packet_id {
276            self.wait_for("PUBACK", |p| {
277                matches!(p, Packet::PubAck(a) if a.packet_id == id).then_some(())
278            })?;
279        }
280        Ok(())
281    }
282
283    /// Subscribe to a topic filter (`+` and `#` allowed) and wait for the SUBACK.
284    pub fn subscribe(&mut self, filter: &str, qos: QoS) -> Result<()> {
285        let id = self.next_packet_id();
286        self.send(&Packet::Subscribe(SubscribePacket {
287            packet_id: id,
288            filters: vec![SubscribeFilter::new(filter, qos)],
289            properties: Properties::default(),
290        }))?;
291        let code = self.wait_for("SUBACK", |p| match p {
292            Packet::SubAck(a) if a.packet_id == id => Some(a.reason_codes.first().copied()),
293            _ => None,
294        })?;
295        match code {
296            Some(c) if c.is_success() => Ok(()),
297            Some(c) => Err(Error::SubscribeRefused(c.0)),
298            None => Err(Error::Unexpected("SUBACK without a reason code")),
299        }
300    }
301
302    /// Unsubscribe from a topic filter and wait for the UNSUBACK.
303    pub fn unsubscribe(&mut self, filter: &str) -> Result<()> {
304        let id = self.next_packet_id();
305        self.send(&Packet::Unsubscribe(UnsubscribePacket {
306            packet_id: id,
307            topic_filters: vec![filter.to_string()],
308            properties: Properties::default(),
309        }))?;
310        self.wait_for("UNSUBACK", |p| {
311            matches!(p, Packet::UnsubAck(a) if a.packet_id == id).then_some(())
312        })
313    }
314
315    /// Wait up to `wait` for a message. Also answers the broker's acknowledgements and sends a
316    /// PINGREQ when the connection has been idle, so call it at least every few seconds.
317    pub fn poll(&mut self, wait: Duration) -> Result<Option<Message>> {
318        if let Some(m) = self.inbox.pop_front() {
319            return Ok(Some(m));
320        }
321        self.keep_alive_tick()?;
322        self.pump(wait)?;
323        self.keep_alive_tick()?;
324        Ok(self.inbox.pop_front())
325    }
326
327    /// Say goodbye (no will message is sent) and give the transport back.
328    pub fn disconnect(mut self) -> Result<T> {
329        self.send(&Packet::Disconnect(DisconnectPacket::normal()))?;
330        Ok(self.io)
331    }
332
333    // ── internals ───────────────────────────────────────────────────────────
334
335    fn next_packet_id(&mut self) -> u16 {
336        self.next_id = self.next_id.checked_add(1).unwrap_or(1);
337        self.next_id
338    }
339
340    fn send(&mut self, packet: &Packet) -> Result<()> {
341        let bytes = packet.encode(self.version)?;
342        self.io.write_all(&bytes)?;
343        self.io.flush()?;
344        self.stats.packets_tx += 1;
345        self.stats.bytes_tx += bytes.len() as u64;
346        self.last_tx = Instant::now();
347        Ok(())
348    }
349
350    /// Send a PINGREQ when idle, fail when the broker does not answer one.
351    fn keep_alive_tick(&mut self) -> Result<()> {
352        if self.keep_alive.is_zero() {
353            return Ok(());
354        }
355        if let Some(sent) = self.ping_sent {
356            if sent.elapsed() > self.keep_alive {
357                return Err(Error::Timeout("PINGRESP"));
358            }
359        } else if self.last_tx.elapsed() >= self.ping_every {
360            self.send(&Packet::PingReq)?;
361            self.ping_sent = Some(Instant::now());
362        }
363        Ok(())
364    }
365
366    /// Read what arrives within `wait`, handle protocol packets, queue messages, and return the
367    /// acknowledgement packets for the caller that is waiting for one.
368    fn pump(&mut self, wait: Duration) -> Result<Vec<Packet>> {
369        self.io
370            .set_read_timeout(Some(wait.max(Duration::from_millis(1))))?;
371        let mut chunk = [0u8; 1024];
372        match self.io.read(&mut chunk) {
373            Ok(0) => return Err(Error::Closed),
374            Ok(n) => {
375                self.stats.bytes_rx += n as u64;
376                self.buf.extend_from_slice(&chunk[..n]);
377            }
378            Err(e)
379                if matches!(
380                    e.kind(),
381                    io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut
382                ) => {}
383            Err(e) => return Err(e.into()),
384        }
385        let mut acks = Vec::new();
386        while let Some(packet) = Packet::decode(&mut self.buf, self.version)? {
387            self.stats.packets_rx += 1;
388            match packet {
389                Packet::Publish(p) => {
390                    self.stats.messages_rx += 1;
391                    let ack = match (p.qos, p.packet_id) {
392                        (QoS::AtLeastOnce, Some(id)) => {
393                            Some(Packet::PubAck(SimpleAck::success(id)))
394                        }
395                        (QoS::ExactlyOnce, Some(id)) => {
396                            Some(Packet::PubRec(SimpleAck::success(id)))
397                        }
398                        _ => None,
399                    };
400                    self.inbox.push_back(Message {
401                        topic: p.topic,
402                        payload: p.payload.to_vec(),
403                        qos: p.qos,
404                        retain: p.retain,
405                    });
406                    if let Some(a) = ack {
407                        self.send(&a)?;
408                    }
409                }
410                Packet::PubRel(a) => {
411                    self.send(&Packet::PubComp(SimpleAck::success(a.packet_id)))?
412                }
413                Packet::PingResp => self.ping_sent = None,
414                Packet::Disconnect(_) => return Err(Error::Closed),
415                other @ (Packet::ConnAck(_)
416                | Packet::PubAck(_)
417                | Packet::SubAck(_)
418                | Packet::UnsubAck(_)) => {
419                    acks.push(other);
420                }
421                _ => {}
422            }
423        }
424        Ok(acks)
425    }
426
427    /// Wait for the first packet `matcher` accepts (messages that arrive meanwhile are queued).
428    fn wait_for<R>(
429        &mut self,
430        what: &'static str,
431        mut matcher: impl FnMut(&Packet) -> Option<R>,
432    ) -> Result<R> {
433        let deadline = Instant::now() + self.ack_timeout;
434        loop {
435            for packet in self.pump(Duration::from_millis(50))? {
436                if let Some(r) = matcher(&packet) {
437                    return Ok(r);
438                }
439            }
440            self.keep_alive_tick()?;
441            if Instant::now() >= deadline {
442                return Err(Error::Timeout(what));
443            }
444        }
445    }
446}
447
448#[cfg(test)]
449mod tests {
450    use super::*;
451    use mqtt_proto::connect::{ConnAckPacket, ConnectReasonCode};
452    use mqtt_proto::subscribe::{SubAckPacket, SubAckReasonCode};
453
454    /// A scripted transport: `incoming` chunks come out one per read, everything written is kept.
455    #[derive(Default)]
456    struct Script {
457        incoming: VecDeque<Vec<u8>>,
458        written: Vec<u8>,
459        closed_after_script: bool,
460    }
461
462    impl Read for Script {
463        fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
464            match self.incoming.pop_front() {
465                Some(chunk) => {
466                    let n = chunk.len().min(buf.len());
467                    buf[..n].copy_from_slice(&chunk[..n]);
468                    if n < chunk.len() {
469                        self.incoming.push_front(chunk[n..].to_vec());
470                    }
471                    Ok(n)
472                }
473                None if self.closed_after_script => Ok(0),
474                None => {
475                    std::thread::sleep(Duration::from_millis(2));
476                    Err(io::ErrorKind::TimedOut.into())
477                }
478            }
479        }
480    }
481
482    impl Write for Script {
483        fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
484            self.written.extend_from_slice(buf);
485            Ok(buf.len())
486        }
487        fn flush(&mut self) -> io::Result<()> {
488            Ok(())
489        }
490    }
491
492    impl Transport for Script {
493        fn set_read_timeout(&mut self, _: Option<Duration>) -> io::Result<()> {
494            Ok(())
495        }
496    }
497
498    fn bytes(p: &Packet) -> Vec<u8> {
499        p.encode(MqttVersion::V311).expect("encodes").to_vec()
500    }
501
502    fn connack(code: u8) -> Vec<u8> {
503        bytes(&Packet::ConnAck(ConnAckPacket {
504            session_present: false,
505            reason_code: ConnectReasonCode::from_v311_return_code(code),
506            properties: Properties::default(),
507        }))
508    }
509
510    fn publish_in(topic: &str, payload: &[u8], qos: QoS, id: Option<u16>) -> Vec<u8> {
511        bytes(&Packet::Publish(PublishPacket {
512            dup: false,
513            qos,
514            retain: false,
515            topic: topic.into(),
516            packet_id: id,
517            payload: Bytes::copy_from_slice(payload),
518            properties: Properties::default(),
519        }))
520    }
521
522    fn connected(extra: Vec<Vec<u8>>) -> Client<Script> {
523        let mut s = Script::default();
524        s.incoming.push_back(connack(0));
525        s.incoming.extend(extra);
526        Client::connect(s, Options::new("test")).expect("connects")
527    }
528
529    fn decode_all(written: &[u8]) -> Vec<Packet> {
530        let mut buf = BytesMut::from(written);
531        let mut out = Vec::new();
532        while let Some(p) = Packet::decode(&mut buf, MqttVersion::V311).expect("decodes") {
533            out.push(p);
534        }
535        out
536    }
537
538    #[test]
539    fn connect_sends_the_options_and_accepts_a_connack() {
540        let mut s = Script::default();
541        s.incoming.push_back(connack(0));
542        let opts = Options::new("c6").credentials("u", "p").will(WillMessage {
543            topic: "dev/status".into(),
544            payload: b"offline".to_vec(),
545            qos: QoS::AtLeastOnce,
546            retain: true,
547        });
548        let client = Client::connect(s, opts).expect("connects");
549        let sent = decode_all(&client.io.written);
550        let Packet::Connect(c) = &sent[0] else {
551            panic!("first packet is CONNECT")
552        };
553        assert_eq!(
554            (c.client_id.as_str(), c.keep_alive, c.clean_start),
555            ("c6", 30, true)
556        );
557        assert_eq!(c.username.as_deref(), Some("u"));
558        assert_eq!(
559            c.will.as_ref().map(|w| w.topic.as_str()),
560            Some("dev/status")
561        );
562        assert_eq!(client.stats().packets_tx, 1);
563        assert_eq!(client.stats().packets_rx, 1);
564    }
565
566    #[test]
567    fn a_refused_connection_says_why() {
568        let mut s = Script::default();
569        s.incoming.push_back(connack(5)); // not authorised
570        assert!(matches!(
571            Client::connect(s, Options::new("x")),
572            Err(Error::Refused(_))
573        ));
574    }
575
576    #[test]
577    fn no_connack_times_out() {
578        let opts = Options {
579            ack_timeout: Duration::from_millis(100),
580            ..Options::new("x")
581        };
582        assert!(matches!(
583            Client::connect(Script::default(), opts),
584            Err(Error::Timeout("CONNACK"))
585        ));
586    }
587
588    #[test]
589    fn qos0_publish_is_one_packet_and_qos2_is_refused() {
590        let mut c = connected(vec![]);
591        c.publish("a/b", b"hi", QoS::AtMostOnce, true)
592            .expect("sent");
593        let sent = decode_all(&c.io.written);
594        let Packet::Publish(p) = &sent[1] else {
595            panic!("a PUBLISH")
596        };
597        assert_eq!(
598            (p.topic.as_str(), &p.payload[..], p.retain, p.packet_id),
599            ("a/b", &b"hi"[..], true, None)
600        );
601        assert_eq!(c.stats().messages_tx, 1);
602        assert!(matches!(
603            c.publish("a", b"x", QoS::ExactlyOnce, false),
604            Err(Error::Unsupported(_))
605        ));
606    }
607
608    #[test]
609    fn qos1_publish_waits_for_its_puback() {
610        let puback = bytes(&Packet::PubAck(SimpleAck::success(1)));
611        let mut c = connected(vec![puback]);
612        c.publish("a", b"x", QoS::AtLeastOnce, false)
613            .expect("acknowledged");
614        // without an answer it times out
615        let mut c2 = connected(vec![]);
616        c2.ack_timeout = Duration::from_millis(80);
617        assert!(matches!(
618            c2.publish("a", b"x", QoS::AtLeastOnce, false),
619            Err(Error::Timeout("PUBACK"))
620        ));
621    }
622
623    #[test]
624    fn subscribe_checks_the_suback_and_keeps_messages_that_arrive_first() {
625        let early = publish_in("t/early", b"1", QoS::AtMostOnce, None);
626        let suback = bytes(&Packet::SubAck(SubAckPacket {
627            packet_id: 1,
628            reason_codes: vec![SubAckReasonCode::granted(QoS::AtLeastOnce)],
629            properties: Properties::default(),
630        }));
631        let mut c = connected(vec![early, suback]);
632        c.subscribe("t/#", QoS::AtLeastOnce).expect("subscribed");
633        let m = c
634            .poll(Duration::from_millis(10))
635            .expect("polls")
636            .expect("the early message");
637        assert_eq!(
638            (m.topic.as_str(), m.payload.as_slice()),
639            ("t/early", &b"1"[..])
640        );
641    }
642
643    #[test]
644    fn a_refused_subscription_is_an_error() {
645        let suback = bytes(&Packet::SubAck(SubAckPacket {
646            packet_id: 1,
647            reason_codes: vec![SubAckReasonCode(0x80)],
648            properties: Properties::default(),
649        }));
650        let mut c = connected(vec![suback]);
651        assert!(matches!(
652            c.subscribe("t", QoS::AtMostOnce),
653            Err(Error::SubscribeRefused(0x80))
654        ));
655    }
656
657    #[test]
658    fn incoming_qos1_is_acknowledged_and_qos2_is_completed() {
659        let mut c = connected(vec![
660            publish_in("a", b"one", QoS::AtLeastOnce, Some(7)),
661            publish_in("b", b"two", QoS::ExactlyOnce, Some(8)),
662            bytes(&Packet::PubRel(SimpleAck::success(8))),
663        ]);
664        let first = c
665            .poll(Duration::from_millis(10))
666            .expect("polls")
667            .expect("message");
668        assert_eq!(first.topic, "a");
669        let second = c
670            .poll(Duration::from_millis(10))
671            .expect("polls")
672            .expect("message");
673        assert_eq!(second.topic, "b");
674        c.poll(Duration::from_millis(10)).expect("polls");
675        let sent = decode_all(&c.io.written);
676        assert!(
677            sent.iter()
678                .any(|p| matches!(p, Packet::PubAck(a) if a.packet_id == 7))
679        );
680        assert!(
681            sent.iter()
682                .any(|p| matches!(p, Packet::PubRec(a) if a.packet_id == 8))
683        );
684        assert!(
685            sent.iter()
686                .any(|p| matches!(p, Packet::PubComp(a) if a.packet_id == 8))
687        );
688        assert_eq!(c.stats().messages_rx, 2);
689    }
690
691    #[test]
692    fn a_packet_split_over_several_reads_is_assembled() {
693        let whole = publish_in("split/topic", b"payload", QoS::AtMostOnce, None);
694        let (a, b) = whole.split_at(5);
695        let (b1, b2) = b.split_at(3);
696        let mut c = connected(vec![a.to_vec(), b1.to_vec(), b2.to_vec()]);
697        let mut got = None;
698        for _ in 0..5 {
699            if let Some(m) = c.poll(Duration::from_millis(5)).expect("polls") {
700                got = Some(m);
701                break;
702            }
703        }
704        assert_eq!(got.expect("assembled").topic, "split/topic");
705    }
706
707    #[test]
708    fn an_idle_connection_is_pinged_and_a_silent_broker_is_a_timeout() {
709        let opts = Options {
710            keep_alive: Duration::from_secs(1),
711            ping_interval: Some(Duration::from_millis(20)),
712            ..Options::new("x")
713        };
714        let mut s = Script::default();
715        s.incoming.push_back(connack(0));
716        let mut c = Client::connect(s, opts).expect("connects");
717        std::thread::sleep(Duration::from_millis(40));
718        c.poll(Duration::from_millis(5)).expect("polls");
719        assert!(
720            decode_all(&c.io.written)
721                .iter()
722                .any(|p| matches!(p, Packet::PingReq)),
723            "a PINGREQ went out"
724        );
725        // the broker never answers: after the keep-alive the connection is declared dead
726        std::thread::sleep(Duration::from_millis(1050));
727        assert!(matches!(
728            c.poll(Duration::from_millis(5)),
729            Err(Error::Timeout("PINGRESP"))
730        ));
731    }
732
733    #[test]
734    fn a_pingresp_clears_the_wait() {
735        let opts = Options {
736            keep_alive: Duration::from_secs(1),
737            ping_interval: Some(Duration::from_millis(20)),
738            ..Options::new("x")
739        };
740        let mut s = Script::default();
741        s.incoming.push_back(connack(0));
742        let mut c = Client::connect(s, opts).expect("connects");
743        std::thread::sleep(Duration::from_millis(40));
744        c.poll(Duration::from_millis(5)).expect("pings");
745        c.io.incoming.push_back(bytes(&Packet::PingResp));
746        c.poll(Duration::from_millis(5)).expect("pong read");
747        assert!(c.ping_sent.is_none());
748    }
749
750    #[test]
751    fn a_closed_connection_is_reported() {
752        let mut c = connected(vec![]);
753        c.io.closed_after_script = true;
754        assert!(matches!(
755            c.poll(Duration::from_millis(5)),
756            Err(Error::Closed)
757        ));
758    }
759
760    #[test]
761    fn a_server_disconnect_is_reported() {
762        let mut c = connected(vec![bytes(&Packet::Disconnect(DisconnectPacket::normal()))]);
763        assert!(matches!(
764            c.poll(Duration::from_millis(5)),
765            Err(Error::Closed)
766        ));
767    }
768
769    #[test]
770    fn garbage_from_the_broker_is_a_protocol_error() {
771        let mut c = connected(vec![vec![0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF]]);
772        assert!(matches!(
773            c.poll(Duration::from_millis(5)),
774            Err(Error::Protocol(_))
775        ));
776    }
777
778    #[test]
779    fn disconnect_sends_the_goodbye_and_returns_the_transport() {
780        let c = connected(vec![]);
781        let io = c.disconnect().expect("sent");
782        assert!(
783            decode_all(&io.written)
784                .iter()
785                .any(|p| matches!(p, Packet::Disconnect(_)))
786        );
787    }
788
789    #[test]
790    fn packet_ids_skip_zero_when_they_wrap() {
791        let mut c = connected(vec![]);
792        c.next_id = u16::MAX;
793        assert_eq!(c.next_packet_id(), 1);
794    }
795}
mqtt_client_blocking/lib.rs

mqtt_client_blocking/
lib.rs
