Add if_ready variants of receive_message and poll, when ReadReady trait is implemented for io

src/client/client.rs

@@ -22,6 +22,7 @@
  * SOFTWARE.
  */
 
+use embedded_io::ReadReady;
 use embedded_io_async::{Read, Write};
 use heapless::Vec;
 use rand_core::RngCore;
@@ -222,3 +223,24 @@ where
         }
     }
 }
+
+impl<'a, T, const MAX_PROPERTIES: usize, R> MqttClient<'a, T, MAX_PROPERTIES, R>
+where
+    T: Read + Write + ReadReady,
+    R: RngCore,
+{
+    /// Receive a message if one is ready. The work of this method strictly depends on the
+    /// network implementation passed in the `ClientConfig`. It expects the PUBLISH packet
+    /// from the broker.
+    pub async fn receive_message_if_ready<'b>(
+        &'b mut self,
+    ) -> Result<Option<(&'b str, &'b [u8])>, ReasonCode> {
+        match self.raw.poll_if_ready::<0>().await? {
+            None => Ok(None),
+            Some(Event::Message(topic, payload)) => Ok(Some((topic, payload))),
+            Some(Event::Disconnect(reason)) => Err(reason),
+            // If an application message comes at this moment, it is lost.
+            _ => Err(ReasonCode::ImplementationSpecificError),
+        }
+    }
+}

src/client/raw_client.rs

@@ -1,3 +1,4 @@
+use embedded_io::ReadReady;
 use embedded_io_async::{Read, Write};
 use heapless::Vec;
 use rand_core::RngCore;
@@ -472,6 +473,29 @@ where
     }
 }
 
+impl<'a, T, const MAX_PROPERTIES: usize, R> RawMqttClient<'a, T, MAX_PROPERTIES, R>
+where
+    T: Read + Write + ReadReady,
+    R: RngCore,
+{
+    pub async fn poll_if_ready<'b, const MAX_TOPICS: usize>(
+        &'b mut self,
+    ) -> Result<Option<Event<'b>>, ReasonCode> {
+        if self.connection.is_none() {
+            return Err(ReasonCode::NetworkError);
+        }
+
+        let conn = self.connection.as_mut().unwrap();
+
+        // If there's no data, just return None
+        if !conn.receive_ready()? {
+            Ok(None)
+        } else {
+            self.poll::<MAX_TOPICS>().await.map(Some)
+        }
+    }
+}
+
 #[cfg(not(feature = "tls"))]
 async fn receive_packet<'c, T: Read + Write>(
     buffer: &mut [u8],

src/network/mod.rs

@@ -23,6 +23,7 @@
  */
 
 use crate::packet::v5::reason_codes::ReasonCode;
+use embedded_io::ReadReady;
 use embedded_io_async::{Read, Write};
 
 pub struct NetworkConnection<T>
@@ -60,3 +61,14 @@ where
             .map_err(|_| ReasonCode::NetworkError)
     }
 }
+
+/// Network connection represents an established TCP connection.
+impl<T> NetworkConnection<T>
+where
+    T: Read + Write + ReadReady,
+{
+    /// Check whether the TCP connection is ready to provide any data
+    pub fn receive_ready(&mut self) -> Result<bool, ReasonCode> {
+        self.io.read_ready().map_err(|_| ReasonCode::NetworkError)
+    }
+}