Repository: quartiq/minimq
Pull request #143 (2023-08-23 08:58:20)

Refactoring configuration to builder semantics
This PR refactors the setup of the Minimq client into a single `Config` struct. This removes the need to do run-time checks on the client state and makes it simpler to see which buffers are being used for what. It also cuts down on the number of arguments needed in `Minimq::new()`


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index f652486..bfd058c 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,6 +7,9 @@ This document describes the changes to Minimq between releases.
 ## Changed
 * [breaking] Const generics for message size and allowable in-flight messages have been removed.
   Instead, the user now supplies an RX buffer, a TX buffer, and a session state buffer.
+ * Setup-only configuration APIs such as `set_will()` and `set_keepalive_interval()` have been moved
+ to a new `Config` structure that is supplied to the `Minimq::new()` constructor to simplify the
+ client.
 
 ## Added
 * Support for subscribing at `QoS::ExactlyOnce`
diff --git a/src/config.rs b/src/config.rs
new file mode 100644
index 0000000..b6abd5e
--- /dev/null
+++ b/src/config.rs
@@ -0,0 +1,81 @@
+use crate::{ProtocolError, Will};
+use core::convert::TryFrom;
+use embedded_time::duration::{Extensions, Milliseconds};
+use heapless::String;
+
+/// Configuration specifying the operational state of the MQTT client.
+pub struct Config<'a, Broker: crate::Broker> {
+    pub(crate) broker: Broker,
+    pub(crate) rx_buffer: &'a mut [u8],
+    pub(crate) tx_buffer: &'a mut [u8],
+    pub(crate) state_buffer: &'a mut [u8],
+    pub(crate) client_id: String<64>,
+    pub(crate) keepalive_interval: Milliseconds<u32>,
+    pub(crate) downgrade_qos: bool,
+    pub(crate) will: Option<Will<'a>>,
+}
+
+impl<'a, Broker: crate::Broker> Config<'a, Broker> {
+    /// Construct configuration for the MQTT client.
+    ///
+    /// # Args
+    /// * `rx` - Memory used for receiving messages. The length of this buffer is the maximum
+    /// receive packet length.
+    /// * `tx` - Memory used for transmitting messages. The length of this buffer is the max
+    /// transmit length.
+    pub fn new(broker: Broker, rx: &'a mut [u8], tx: &'a mut [u8]) -> Self {
+        Self {
+            broker,
+            rx_buffer: rx,
+            tx_buffer: tx,
+            state_buffer: &mut [],
+            client_id: String::new(),
+            keepalive_interval: 59_000.milliseconds(),
+            downgrade_qos: false,
+            will: None,
+        }
+    }
+
+    /// Provide additional buffer space if messages above [QoS::AtMostOnce] are required.
+    pub fn session_state(mut self, buffer: &'a mut [u8]) -> Self {
+        self.state_buffer = buffer;
+        self
+    }
+
+    /// Specify a known client ID to use. If not assigned, the broker will auto assign an ID.
+    pub fn client_id(mut self, id: &str) -> Result<Self, ProtocolError> {
+        self.client_id =
+            String::try_from(id).map_err(|_| ProtocolError::ProvidedClientIdTooLong)?;
+        Ok(self)
+    }
+
+    /// Configure the MQTT keep-alive interval.
+    ///
+    /// # Note
+    /// The broker may override the requested keep-alive interval. Any value requested by the
+    /// broker will be used instead.
+    ///
+    /// # Args
+    /// * `interval` - The keep-alive interval in seconds. A ping will be transmitted if no other
+    /// messages are sent within 50% of the keep-alive interval.
+    pub fn keepalive_interval(mut self, seconds: u16) -> Self {
+        self.keepalive_interval = Milliseconds(seconds as u32 * 1000);
+        self
+    }
+
+    /// Specify if publication [QoS] should be automatically downgraded to the maximum supported by
+    /// the server if they exceed the server [QoS] maximum.
+    pub fn autodowngrade_qos(mut self) -> Self {
+        self.downgrade_qos = true;
+        self
+    }
+
+    /// Specify the Will message to be sent if the client disconnects.
+    ///
+    /// # Args
+    /// * `will` - The will to use.
+    pub fn will(mut self, will: Will<'a>) -> Self {
+        self.will.replace(will);
+        self
+    }
+}
diff --git a/src/lib.rs b/src/lib.rs
index c87382f..9992204 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -23,7 +23,7 @@
 //! Below is a sample snippet showing how this library is used.
 //!
 //! ```no_run
-//! use minimq::{Minimq, Publication};
+//! use minimq::{Config, Minimq, Publication};
 //!
 //! // Construct an MQTT client with a maximum packet size of 256 bytes
 //! // and a maximum of 16 messages that are allowed to be "in flight".
@@ -35,13 +35,12 @@
 //! let mut session = [0; 256];
 //! let localhost: embedded_nal::IpAddr = "127.0.0.1".parse().unwrap();
 //! let mut mqtt: Minimq<'_, _, _, minimq::broker::IpBroker> = Minimq::new(
-//!         localhost.into(),
-//!         "test",
 //!         std_embedded_nal::Stack::default(),
 //!         std_embedded_time::StandardClock::default(),
-//!         &mut rx_buffer,
-//!         &mut tx_buffer,
-//!         &mut session).unwrap();
+//!         Config::new(localhost.into(), &mut rx_buffer, &mut tx_buffer)
+//!             .session_state(&mut session)
+//!             .client_id("test").unwrap(),
+//!         );
 //!
 //! let mut subscribed = false;
 //!
@@ -66,11 +65,8 @@
 //! ```
 
 pub mod broker;
+pub mod config;
 mod de;
-mod ser;
-
-pub use broker::Broker;
-
 mod message_types;
 pub mod mqtt_client;
 mod network_manager;
@@ -80,11 +76,14 @@ pub mod publication;
 mod reason_codes;
 mod republication;
 mod ring_buffer;
+mod ser;
 mod session_state;
 pub mod types;
 mod varint;
 mod will;
 
+pub use broker::Broker;
+pub use config::Config;
 pub use properties::Property;
 pub use publication::Publication;
 pub use reason_codes::ReasonCode;
diff --git a/src/mqtt_client.rs b/src/mqtt_client.rs
index 0531bc8..90f3b24 100644
--- a/src/mqtt_client.rs
+++ b/src/mqtt_client.rs
@@ -6,12 +6,12 @@ use crate::{
     session_state::SessionState,
     types::{Properties, TopicFilter, Utf8String},
     will::Will,
-    Error, MinimqError, Property, ProtocolError, QoS, {debug, error, info, warn},
+    Config, Error, MinimqError, Property, ProtocolError, QoS, {debug, error, info, warn},
 };
 
 use core::convert::{TryFrom, TryInto};
 use embedded_time::{
-    duration::{Extensions, Milliseconds, Seconds},
+    duration::{Milliseconds, Seconds},
     Instant,
 };
 
@@ -54,7 +54,7 @@ struct ClientContext<'a, Clock: embedded_time::Clock> {
     send_quota: u16,
     max_send_quota: u16,
     maximum_packet_size: Option<u32>,
-    keep_alive_interval: Option<Milliseconds<u32>>,
+    keep_alive_interval: Milliseconds<u32>,
     pending_subscriptions: heapless::Vec<u16, 32>,
 
     ping_timeout: Option<Instant<Clock>>,
@@ -67,7 +67,11 @@ impl<'a, Clock> ClientContext<'a, Clock>
 where
     Clock: embedded_time::Clock,
 {
-    pub fn new(clock: Clock, session_state: SessionState<'a>) -> Self {
+    pub fn new(
+        clock: Clock,
+        session_state: SessionState<'a>,
+        keepalive: Milliseconds<u32>,
+    ) -> Self {
         Self {
             session_state,
             send_quota: u16::MAX,
@@ -77,7 +81,7 @@ where
             ping_timeout: None,
             next_ping: None,
             max_qos: None,
-            keep_alive_interval: Some(59_000.milliseconds()),
+            keep_alive_interval: keepalive,
             maximum_packet_size: None,
         }
     }
@@ -111,9 +115,8 @@ where
         self.ping_timeout = None;
 
         // The next ping should be sent out in half the keep-alive interval from now.
-        if let Some(interval) = self.keep_alive_interval {
-            self.next_ping.replace(self.clock.try_now()? + interval / 2);
-        }
+        self.next_ping
+            .replace(self.clock.try_now()? + self.keep_alive_interval / 2);
 
         Ok(())
     }
@@ -147,20 +150,17 @@ where
 
         let now = self.clock.try_now()?;
 
-        Ok(self
-            .keep_alive_interval
-            .zip(self.next_ping)
-            .map(|(keep_alive_interval, ping_deadline)| {
-                // Update the next ping deadline if the ping is due.
-                if now > ping_deadline {
-                    // The next ping should be sent out in half the keep-alive interval from now.
-                    self.next_ping.replace(now + keep_alive_interval / 2);
-                    self.ping_timeout.replace(now + PING_TIMEOUT);
-                }
+        let Some(ping_deadline) = self.next_ping else {
+            return Ok(false);
+        };
+        // Update the next ping deadline if the ping is due.
+        if now > ping_deadline {
+            // The next ping should be sent out in half the keep-alive interval from now.
+            self.next_ping.replace(now + self.keep_alive_interval / 2);
+            self.ping_timeout.replace(now + PING_TIMEOUT);
+        }
 
-                now > ping_deadline
-            })
-            .unwrap_or(false))
+        Ok(now > ping_deadline)
     }
 
     /// Get the keep-alive interval as an integer number of seconds.
@@ -168,20 +168,7 @@ where
     /// # Note
     /// If no keep-alive interval is specified, zero is returned.
     pub fn keepalive_interval(&self) -> u16 {
-        (self
-            .keep_alive_interval
-            .unwrap_or_else(|| 0.milliseconds())
-            .0
-            / 1000) as u16
-    }
-
-    /// Update the keep-alive interval.
-    ///
-    /// # Args
-    /// * `seconds` - The number of seconds in the keep-alive interval.
-    pub fn set_keepalive(&mut self, seconds: u16) {
-        self.keep_alive_interval
-            .replace(Milliseconds(seconds as u32 * 1000));
+        (self.keep_alive_interval.0 / 1000) as u16
     }
 }
 
@@ -244,7 +231,7 @@ where
                         String::from_str(id.0).or(Err(ProtocolError::ProvidedClientIdTooLong))?;
                 }
                 Property::ServerKeepAlive(keep_alive) => {
-                    self.set_keepalive(keep_alive);
+                    self.keep_alive_interval = Milliseconds(keep_alive as u32 * 1000);
                 }
                 Property::ReceiveMaximum(max) => {
                     self.send_quota = max.max(self.session_state.max_send_quota());
@@ -290,55 +277,6 @@ pub struct MqttClient<
 impl<'buf, TcpStack: TcpClientStack, Clock: embedded_time::Clock, Broker: crate::Broker>
     MqttClient<'buf, TcpStack, Clock, Broker>
 {
-    /// Specify the Will message to be sent if the client disconnects.
-    ///
-    /// # Args
-    /// * `will` - The will to use.
-    pub fn set_will(&mut self, will: Will<'buf>) -> Result<(), Error<TcpStack::Error>> {
-        self.will.replace(will);
-        Ok(())
-    }
-
-    /// Configure the MQTT keep-alive interval.
-    ///
-    /// # Note
-    /// This must be completed before connecting to a broker.
-    ///
-    /// # Note
-    /// The broker may override the requested keep-alive interval. Any value requested by the
-    /// broker will be used instead.
-    ///
-    /// # Args
-    /// * `interval` - The keep-alive interval in seconds. A ping will be transmitted if no other
-    /// messages are sent within 50% of the keep-alive interval.
-    pub fn set_keepalive_interval(
-        &mut self,
-        interval_seconds: u16,
-    ) -> Result<(), Error<TcpStack::Error>> {
-        if self.sm.state() != &States::Disconnected {
-            return Err(Error::NotReady);
-        }
-
-        self.sm.context_mut().set_keepalive(interval_seconds);
-        Ok(())
-    }
-
-    /// Set custom MQTT port to connect to.
-    ///
-    /// # Note
-    /// This must be completed before connecting to a broker.
-    ///
-    /// # Args
-    /// * `port` - The Port number to connect to.
-    pub fn set_broker_port(&mut self, port: u16) -> Result<(), Error<TcpStack::Error>> {
-        if self.sm.state() != &States::Disconnected {
-            return Err(Error::NotReady);
-        }
-
-        self.broker.set_port(port);
-        Ok(())
-    }
-
     /// Subscribe to a topic.
     ///
     /// # Note
@@ -386,12 +324,6 @@ impl<'buf, TcpStack: TcpClientStack, Clock: embedded_time::Clock, Broker: crate:
         !self.sm.context().pending_subscriptions.is_empty()
     }
 
-    /// Specify if publication [QoS] should be automatically downgraded to the maximum supported by
-    /// the server if they exceed the server [QoS] maximum.
-    pub fn autodowngrade_qos(&mut self, enabled: bool) {
-        self.downgrade_qos = enabled;
-    }
-
     /// Determine if the client has established a connection with the broker.
     ///
     /// # Returns
@@ -758,32 +690,28 @@ impl<'buf, TcpStack: TcpClientStack, Clock: embedded_time::Clock, Broker: crate:
     /// # Returns
     /// A `Minimq` object that can be used for publishing messages, subscribing to topics, and
     /// managing the MQTT state.
-    pub fn new(
-        broker: Broker,
-        client_id: &str,
-        network_stack: TcpStack,
-        clock: Clock,
-        rx_buffer: &'buf mut [u8],
-        tx_buffer: &'buf mut [u8],
-        state_buffer: &'buf mut [u8],
-    ) -> Result<Self, Error<TcpStack::Error>> {
-        let client_id =
-            String::from_str(client_id).or(Err(ProtocolError::ProvidedClientIdTooLong))?;
-        let session_state = SessionState::new(client_id, state_buffer, tx_buffer.len());
-
-        let minimq = Minimq {
+    pub fn new(network_stack: TcpStack, clock: Clock, config: Config<'buf, Broker>) -> Self {
+        let session_state = SessionState::new(
+            config.client_id,
+            config.state_buffer,
+            config.tx_buffer.len(),
+        );
+
+        Minimq {
             client: MqttClient {
-                sm: StateMachine::new(ClientContext::new(clock, session_state)),
-                downgrade_qos: false,
-                broker,
-                will: None,
-                network: InterfaceHolder::new(network_stack, tx_buffer),
-                max_packet_size: rx_buffer.len(),
+                sm: StateMachine::new(ClientContext::new(
+                    clock,
+                    session_state,
+                    config.keepalive_interval,
+                )),
+                downgrade_qos: config.downgrade_qos,
+                broker: config.broker,
+                will: config.will,
+                network: InterfaceHolder::new(network_stack, config.tx_buffer),
+                max_packet_size: config.rx_buffer.len(),
             },
-            packet_reader: PacketReader::new(rx_buffer),
-        };
-
-        Ok(minimq)
+            packet_reader: PacketReader::new(config.rx_buffer),
+        }
     }
 
     /// Check the MQTT interface for available messages.


Test patch:
diff --git a/tests/at_least_once_subscription.rs b/tests/at_least_once_subscription.rs
index 76f5cd6..3e74bdc 100644
--- a/tests/at_least_once_subscription.rs
+++ b/tests/at_least_once_subscription.rs
@@ -16,18 +16,12 @@ fn main() -> std::io::Result<()> {
     let stack = std_embedded_nal::Stack::default();
     let localhost = IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1));
     let mut mqtt: Minimq<'_, _, _, minimq::broker::IpBroker> = Minimq::new(
-        localhost.into(),
-        "",
         stack,
         StandardClock::default(),
-        &mut rx_buffer,
-        &mut tx_buffer,
-        &mut session,
-    )
-    .unwrap();
-
-    // Use a keepalive interval for the client.
-    mqtt.client().set_keepalive_interval(60).unwrap();
+        minimq::Config::new(localhost.into(), &mut rx_buffer, &mut tx_buffer)
+            .session_state(&mut session)
+            .keepalive_interval(60),
+    );
 
     let mut published = false;
     let mut subscribed = false;
diff --git a/tests/exactly_once_subscription.rs b/tests/exactly_once_subscription.rs
index c74d711..60d3b80 100644
--- a/tests/exactly_once_subscription.rs
+++ b/tests/exactly_once_subscription.rs
@@ -16,18 +16,12 @@ fn main() -> std::io::Result<()> {
     let stack = std_embedded_nal::Stack::default();
     let localhost = IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1));
     let mut mqtt: Minimq<'_, _, _, minimq::broker::IpBroker> = Minimq::new(
-        localhost.into(),
-        "",
         stack,
         StandardClock::default(),
-        &mut rx_buffer,
-        &mut tx_buffer,
-        &mut session,
-    )
-    .unwrap();
-
-    // Use a keepalive interval for the client.
-    mqtt.client().set_keepalive_interval(60).unwrap();
+        minimq::Config::new(localhost.into(), &mut rx_buffer, &mut tx_buffer)
+            .session_state(&mut session)
+            .keepalive_interval(60),
+    );
 
     let mut published = false;
     let mut subscribed = false;
diff --git a/tests/exactly_once_transmission.rs b/tests/exactly_once_transmission.rs
index 3a76ff2..4190635 100644
--- a/tests/exactly_once_transmission.rs
+++ b/tests/exactly_once_transmission.rs
@@ -13,18 +13,12 @@ fn main() -> std::io::Result<()> {
     let stack = std_embedded_nal::Stack::default();
     let localhost = IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1));
     let mut mqtt: Minimq<'_, _, _, minimq::broker::IpBroker> = Minimq::new(
-        localhost.into(),
-        "",
         stack,
         StandardClock::default(),
-        &mut rx_buffer,
-        &mut tx_buffer,
-        &mut session,
-    )
-    .unwrap();
-
-    // Use a keepalive interval for the client.
-    mqtt.client().set_keepalive_interval(60).unwrap();
+        minimq::Config::new(localhost.into(), &mut rx_buffer, &mut tx_buffer)
+            .session_state(&mut session)
+            .keepalive_interval(60),
+    );
 
     let mut published = false;
 
diff --git a/tests/integration_test.rs b/tests/integration_test.rs
index 447d93d..26157b9 100644
--- a/tests/integration_test.rs
+++ b/tests/integration_test.rs
@@ -7,32 +7,26 @@ use std_embedded_time::StandardClock;
 fn main() -> std::io::Result<()> {
     env_logger::init();
 
+    let will = Will::new("exit", "Test complete".as_bytes(), &[]).unwrap();
+
     let mut rx_buffer = [0u8; 256];
     let mut tx_buffer = [0u8; 256];
     let mut session = [0u8; 256];
     let stack = std_embedded_nal::Stack::default();
     let localhost = IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1));
     let mut mqtt: Minimq<'_, _, _, minimq::broker::IpBroker> = Minimq::new(
-        localhost.into(),
-        "",
         stack,
         StandardClock::default(),
-        &mut rx_buffer,
-        &mut tx_buffer,
-        &mut session,
-    )
-    .unwrap();
-
-    // Use a keepalive interval for the client.
-    mqtt.client().set_keepalive_interval(60).unwrap();
+        minimq::Config::new(localhost.into(), &mut rx_buffer, &mut tx_buffer)
+            .session_state(&mut session)
+            .will(will)
+            .keepalive_interval(60),
+    );
 
     let mut published = false;
     let mut subscribed = false;
     let mut responses = 0;
 
-    let will = Will::new("exit", "Test complete".as_bytes(), &[]).unwrap();
-    mqtt.client().set_will(will).unwrap();
-
     loop {
         // Continually poll the client until there is no more data.
         while let Some(was_response) = mqtt
diff --git a/tests/poll_result.rs b/tests/poll_result.rs
index 813a7ef..97d5beb 100644
--- a/tests/poll_result.rs
+++ b/tests/poll_result.rs
@@ -13,18 +13,12 @@ fn main() -> std::io::Result<()> {
     let stack = std_embedded_nal::Stack::default();
     let localhost = IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1));
     let mut mqtt: Minimq<'_, _, _, minimq::broker::IpBroker> = Minimq::new(
-        localhost.into(),
-        "",
         stack,
         StandardClock::default(),
-        &mut rx_buffer,
-        &mut tx_buffer,
-        &mut session,
-    )
-    .unwrap();
-
-    // Use a keepalive interval for the client.
-    mqtt.client().set_keepalive_interval(60).unwrap();
+        minimq::Config::new(localhost.into(), &mut rx_buffer, &mut tx_buffer)
+            .session_state(&mut session)
+            .keepalive_interval(60),
+    );
 
     let mut published = false;
     let mut subscribed = false;
diff --git a/tests/reconnection.rs b/tests/reconnection.rs
index d23b1e6..198c021 100644
--- a/tests/reconnection.rs
+++ b/tests/reconnection.rs
@@ -19,18 +19,12 @@ fn main() -> std::io::Result<()> {
     let stack = stack::MitmStack::new(&sockets);
     let localhost = IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1));
     let mut mqtt: Minimq<'_, _, _, minimq::broker::IpBroker> = Minimq::new(
-        localhost.into(),
-        "",
         stack,
         StandardClock::default(),
-        &mut rx_buffer,
-        &mut tx_buffer,
-        &mut session,
-    )
-    .unwrap();
-
-    // Use a keepalive interval for the client.
-    mqtt.client().set_keepalive_interval(1).unwrap();
+        minimq::Config::new(localhost.into(), &mut rx_buffer, &mut tx_buffer)
+            .session_state(&mut session)
+            .keepalive_interval(1),
+    );
 
     // 1. Poll until we're connected and subscribed to a test topic
     while !mqtt.client().is_connected() {