Repository: quartiq/minimq (rust)

Issue:
Add support for domain-name brokers
Add support for accepting a domain name for the broker.

When a domain name is provided, the name must first be resolved into an IP address before being used for connections.

Pull request description:
Adding support for domain-named brokers
Adds support for using brokers by domain name.

Fixes #120 

TODO:
- [x] Test with a DNS named broker locally
- [x] Document the broker configurations

cc @jordens 

Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index d3de53b..d7fe489 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -10,6 +10,7 @@ This document describes the changes to Minimq between releases.
 
 ## Added
 * Support for subscribing at `QoS::ExactlyOnce`
+* Brokers may now be provided using domain-name syntax or static IP addresses.
 
 ## Fixed
 * Fixed an issue where PubComp was serialized with an incorrect control code
diff --git a/src/broker.rs b/src/broker.rs
new file mode 100644
index 0000000..1d78526
--- /dev/null
+++ b/src/broker.rs
@@ -0,0 +1,92 @@
+use crate::{warn, MQTT_INSECURE_DEFAULT_PORT};
+use core::convert::TryFrom;
+use embedded_nal::{nb, AddrType, Dns, IpAddr, Ipv4Addr, SocketAddr};
+
+/// A type that allows us to (eventually) determine the broker address.
+pub trait Broker {
+    /// Retrieve the broker address (if available).
+    fn get_address(&mut self) -> Option<SocketAddr>;
+
+    /// Set the port of the broker connection.
+    fn set_port(&mut self, port: u16);
+}
+
+/// A broker that is specified using a qualified domain-name. The name will be resolved at some
+/// point in the future.
+pub struct NamedBroker<R: Dns, const T: usize = 253> {
+    raw: heapless::String<T>,
+    resolver: R,
+    addr: SocketAddr,
+}
+
+impl<R: Dns, const T: usize> NamedBroker<R, T> {
+    /// Construct a new named broker.
+    ///
+    /// # Args
+    /// * `broker` - The domain name of the broker, such as `broker.example.com`
+    /// * `resolver` - A [embedded_nal::Dns] resolver to resolve the broker domain name to an IP
+    /// address.
+    pub fn new(broker: &str, resolver: R) -> Result<Self, &str> {
+        let addr: Ipv4Addr = broker.parse().unwrap_or(Ipv4Addr::UNSPECIFIED);
+
+        Ok(Self {
+            raw: heapless::String::try_from(broker).map_err(|_| "Broker domain name too long")?,
+            resolver,
+            addr: SocketAddr::new(IpAddr::V4(addr), MQTT_INSECURE_DEFAULT_PORT),
+        })
+    }
+}
+
+impl<R: Dns, const T: usize> Broker for NamedBroker<R, T> {
+    fn get_address(&mut self) -> Option<SocketAddr> {
+        // Attempt to resolve the address.
+        if self.addr.ip().is_unspecified() {
+            match self.resolver.get_host_by_name(&self.raw, AddrType::IPv4) {
+                Ok(ip) => self.addr.set_ip(ip),
+                Err(nb::Error::WouldBlock) => {}
+                Err(_other) => {
+                    warn!("DNS lookup failed: {_other:?}")
+                }
+            }
+        }
+
+        if !self.addr.ip().is_unspecified() {
+            Some(self.addr)
+        } else {
+            None
+        }
+    }
+
+    fn set_port(&mut self, port: u16) {
+        self.addr.set_port(port)
+    }
+}
+
+/// A simple broker specification where the address of the broker is known in advance.
+pub struct IpBroker {
+    addr: SocketAddr,
+}
+
+impl IpBroker {
+    /// Construct a broker with a known IP address.
+    pub fn new(broker: IpAddr) -> Self {
+        Self {
+            addr: SocketAddr::new(broker, MQTT_INSECURE_DEFAULT_PORT),
+        }
+    }
+}
+impl Broker for IpBroker {
+    fn get_address(&mut self) -> Option<SocketAddr> {
+        Some(self.addr)
+    }
+
+    fn set_port(&mut self, port: u16) {
+        self.addr.set_port(port)
+    }
+}
+
+impl From<IpAddr> for IpBroker {
+    fn from(addr: IpAddr) -> Self {
+        IpBroker::new(addr)
+    }
+}
diff --git a/src/lib.rs b/src/lib.rs
index e7115d2..c87382f 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -33,8 +33,9 @@
 //! let mut rx_buffer = [0; 256];
 //! let mut tx_buffer = [0; 256];
 //! let mut session = [0; 256];
-//! let mut mqtt = Minimq::new(
-//!         "127.0.0.1".parse().unwrap(),
+//! 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(),
@@ -64,9 +65,12 @@
 //! }
 //! ```
 
+pub mod broker;
 mod de;
 mod ser;
 
+pub use broker::Broker;
+
 mod message_types;
 pub mod mqtt_client;
 mod network_manager;
diff --git a/src/mqtt_client.rs b/src/mqtt_client.rs
index fc44d7c..54009eb 100644
--- a/src/mqtt_client.rs
+++ b/src/mqtt_client.rs
@@ -6,8 +6,7 @@ use crate::{
     session_state::SessionState,
     types::{Properties, TopicFilter, Utf8String},
     will::Will,
-    Error, MinimqError, Property, ProtocolError, QoS, MQTT_INSECURE_DEFAULT_PORT,
-    {debug, error, info, warn},
+    Error, MinimqError, Property, ProtocolError, QoS, {debug, error, info, warn},
 };
 
 use core::convert::TryInto;
@@ -16,7 +15,7 @@ use embedded_time::{
     Instant,
 };
 
-use embedded_nal::{IpAddr, SocketAddr, TcpClientStack};
+use embedded_nal::TcpClientStack;
 
 use heapless::String;
 
@@ -268,16 +267,21 @@ impl<E> From<sm::Error<MinimqError>> for Error<E> {
 }
 
 /// The client that can be used for interacting with the MQTT broker.
-pub struct MqttClient<'buf, TcpStack: TcpClientStack, Clock: embedded_time::Clock> {
+pub struct MqttClient<
+    'buf,
+    TcpStack: TcpClientStack,
+    Clock: embedded_time::Clock,
+    Broker: crate::Broker,
+> {
     sm: sm::StateMachine<ClientContext<'buf, Clock>>,
     network: InterfaceHolder<'buf, TcpStack>,
     will: Option<Will<'buf>>,
-    broker: SocketAddr,
+    broker: Broker,
     max_packet_size: usize,
 }
 
-impl<'buf, TcpStack: TcpClientStack, Clock: embedded_time::Clock>
-    MqttClient<'buf, TcpStack, Clock>
+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.
     ///
@@ -538,9 +542,11 @@ impl<'buf, TcpStack: TcpClientStack, Clock: embedded_time::Clock>
 
         match *self.sm.state() {
             States::Disconnected => {
-                if self.network.connect(self.broker)? {
-                    info!("TCP socket connected");
-                    self.sm.process_event(Events::TcpConnected)?;
+                if let Some(broker) = self.broker.get_address() {
+                    if self.network.connect(broker)? {
+                        info!("TCP socket connected");
+                        self.sm.process_event(Events::TcpConnected)?;
+                    }
                 }
             }
             States::Restart => self.handle_restart()?,
@@ -557,7 +563,12 @@ impl<'buf, TcpStack: TcpClientStack, Clock: embedded_time::Clock>
         f: &mut F,
     ) -> Result<Option<T>, Error<TcpStack::Error>>
     where
-        F: FnMut(&mut MqttClient<'buf, TcpStack, Clock>, &'a str, &[u8], &Properties<'a>) -> T,
+        F: FnMut(
+            &mut MqttClient<'buf, TcpStack, Clock, Broker>,
+            &'a str,
+            &[u8],
+            &Properties<'a>,
+        ) -> T,
     {
         match control_packet {
             ReceivedPacket::ConnAck(ack) => {
@@ -702,20 +713,24 @@ impl<'buf, TcpStack: TcpClientStack, Clock: embedded_time::Clock>
 /// # Note
 /// To connect and maintain an MQTT connection, the `Minimq::poll()` method must be called
 /// regularly.
-pub struct Minimq<'buf, TcpStack, Clock>
+pub struct Minimq<'buf, TcpStack, Clock, Broker>
 where
     TcpStack: TcpClientStack,
     Clock: embedded_time::Clock,
+    Broker: crate::Broker,
 {
-    client: MqttClient<'buf, TcpStack, Clock>,
+    client: MqttClient<'buf, TcpStack, Clock, Broker>,
     packet_reader: PacketReader<'buf>,
 }
 
-impl<'buf, TcpStack: TcpClientStack, Clock: embedded_time::Clock> Minimq<'buf, TcpStack, Clock> {
+impl<'buf, TcpStack: TcpClientStack, Clock: embedded_time::Clock, Broker: crate::Broker>
+    Minimq<'buf, TcpStack, Clock, Broker>
+{
     /// Construct a new MQTT interface.
     ///
     /// # Args
-    /// * `broker` - The IP address of the broker to connect to.
+    /// * `broker` - The broker to connect to. See [crate::broker::NamedBroker] and
+    /// [crate::broker::IpBroker].
     /// * `client_id` The client ID to use for communicating with the broker. If empty, rely on the
     ///   broker to automatically assign a client ID.
     /// * `network_stack` - The network stack to use for communication.
@@ -725,7 +740,7 @@ impl<'buf, TcpStack: TcpClientStack, Clock: embedded_time::Clock> Minimq<'buf, T
     /// A `Minimq` object that can be used for publishing messages, subscribing to topics, and
     /// managing the MQTT state.
     pub fn new(
-        broker: IpAddr,
+        broker: Broker,
         client_id: &str,
         network_stack: TcpStack,
         clock: Clock,
@@ -740,7 +755,7 @@ impl<'buf, TcpStack: TcpClientStack, Clock: embedded_time::Clock> Minimq<'buf, T
         let minimq = Minimq {
             client: MqttClient {
                 sm: StateMachine::new(ClientContext::new(clock, session_state)),
-                broker: SocketAddr::new(broker, MQTT_INSECURE_DEFAULT_PORT),
+                broker,
                 will: None,
                 network: InterfaceHolder::new(network_stack, tx_buffer),
                 max_packet_size: rx_buffer.len(),
@@ -773,8 +788,12 @@ impl<'buf, TcpStack: TcpClientStack, Clock: embedded_time::Clock> Minimq<'buf, T
     /// subscriptions or session state.
     pub fn poll<F, T>(&mut self, mut f: F) -> Result<Option<T>, Error<TcpStack::Error>>
     where
-        for<'a> F:
-            FnMut(&mut MqttClient<'buf, TcpStack, Clock>, &'a str, &[u8], &Properties<'a>) -> T,
+        for<'a> F: FnMut(
+            &mut MqttClient<'buf, TcpStack, Clock, Broker>,
+            &'a str,
+            &[u8],
+            &Properties<'a>,
+        ) -> T,
     {
         self.client.update()?;
 
@@ -822,7 +841,7 @@ impl<'buf, TcpStack: TcpClientStack, Clock: embedded_time::Clock> Minimq<'buf, T
     }
 
     /// Directly access the MQTT client.
-    pub fn client(&mut self) -> &mut MqttClient<'buf, TcpStack, Clock> {
+    pub fn client(&mut self) -> &mut MqttClient<'buf, TcpStack, Clock, Broker> {
         &mut self.client
     }
 }


Test patch:
diff --git a/tests/at_least_once_subscription.rs b/tests/at_least_once_subscription.rs
index c7ad535..76f5cd6 100644
--- a/tests/at_least_once_subscription.rs
+++ b/tests/at_least_once_subscription.rs
@@ -15,8 +15,8 @@ fn main() -> std::io::Result<()> {
     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::new(
-        localhost,
+    let mut mqtt: Minimq<'_, _, _, minimq::broker::IpBroker> = Minimq::new(
+        localhost.into(),
         "",
         stack,
         StandardClock::default(),
diff --git a/tests/exactly_once_subscription.rs b/tests/exactly_once_subscription.rs
index 762fcef..c74d711 100644
--- a/tests/exactly_once_subscription.rs
+++ b/tests/exactly_once_subscription.rs
@@ -15,8 +15,8 @@ fn main() -> std::io::Result<()> {
     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::new(
-        localhost,
+    let mut mqtt: Minimq<'_, _, _, minimq::broker::IpBroker> = Minimq::new(
+        localhost.into(),
         "",
         stack,
         StandardClock::default(),
diff --git a/tests/exactly_once_transmission.rs b/tests/exactly_once_transmission.rs
index 1e064ba..3a76ff2 100644
--- a/tests/exactly_once_transmission.rs
+++ b/tests/exactly_once_transmission.rs
@@ -12,8 +12,8 @@ fn main() -> std::io::Result<()> {
     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::new(
-        localhost,
+    let mut mqtt: Minimq<'_, _, _, minimq::broker::IpBroker> = Minimq::new(
+        localhost.into(),
         "",
         stack,
         StandardClock::default(),
diff --git a/tests/integration_test.rs b/tests/integration_test.rs
index 1df64cf..447d93d 100644
--- a/tests/integration_test.rs
+++ b/tests/integration_test.rs
@@ -12,8 +12,8 @@ fn main() -> std::io::Result<()> {
     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::new(
-        localhost,
+    let mut mqtt: Minimq<'_, _, _, minimq::broker::IpBroker> = Minimq::new(
+        localhost.into(),
         "",
         stack,
         StandardClock::default(),
diff --git a/tests/poll_result.rs b/tests/poll_result.rs
index 15ab845..813a7ef 100644
--- a/tests/poll_result.rs
+++ b/tests/poll_result.rs
@@ -12,8 +12,8 @@ fn main() -> std::io::Result<()> {
     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::new(
-        localhost,
+    let mut mqtt: Minimq<'_, _, _, minimq::broker::IpBroker> = Minimq::new(
+        localhost.into(),
         "",
         stack,
         StandardClock::default(),
diff --git a/tests/reconnection.rs b/tests/reconnection.rs
index e737244..d23b1e6 100644
--- a/tests/reconnection.rs
+++ b/tests/reconnection.rs
@@ -18,8 +18,8 @@ fn main() -> std::io::Result<()> {
     let sockets = std::cell::RefCell::new(Vec::new());
     let stack = stack::MitmStack::new(&sockets);
     let localhost = IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1));
-    let mut mqtt = Minimq::new(
-        localhost,
+    let mut mqtt: Minimq<'_, _, _, minimq::broker::IpBroker> = Minimq::new(
+        localhost.into(),
         "",
         stack,
         StandardClock::default(),