Support embedded_io_async::ReadReady trait

src/client/client.rs

@@ -22,7 +22,7 @@
  * SOFTWARE.
  */
 
-use embedded_io_async::{Read, Write};
+use embedded_io_async::{Read, ReadReady, Write};
 use heapless::Vec;
 use rand_core::RngCore;
 
@@ -34,14 +34,14 @@ use super::raw_client::{Event, RawMqttClient};
 
 pub struct MqttClient<'a, T, const MAX_PROPERTIES: usize, R: RngCore>
 where
-    T: Read + Write,
+    T: Read + ReadReady + Write,
 {
     raw: RawMqttClient<'a, T, MAX_PROPERTIES, R>,
 }
 
 impl<'a, T, const MAX_PROPERTIES: usize, R> MqttClient<'a, T, MAX_PROPERTIES, R>
 where
-    T: Read + Write,
+    T: Read + ReadReady + Write,
     R: RngCore,
 {
     pub fn new(
@@ -221,4 +221,9 @@ where
             _ => Err(ReasonCode::ImplementationSpecificError),
         }
     }
+
+    /// Get whether a TCP connection reader is ready.
+    pub fn read_ready(&mut self) -> Result<bool, ReasonCode> {
+        self.raw.read_ready()
+    }
 }

src/client/raw_client.rs

@@ -1,4 +1,4 @@
-use embedded_io_async::{Read, Write};
+use embedded_io_async::{Read, ReadReady, Write};
 use heapless::Vec;
 use rand_core::RngCore;
 
@@ -38,7 +38,7 @@ pub enum Event<'a> {
 
 pub struct RawMqttClient<'a, T, const MAX_PROPERTIES: usize, R: RngCore>
 where
-    T: Read + Write,
+    T: Read + ReadReady + Write,
 {
     connection: Option<NetworkConnection<T>>,
     buffer: &'a mut [u8],
@@ -50,7 +50,7 @@ where
 
 impl<'a, T, const MAX_PROPERTIES: usize, R> RawMqttClient<'a, T, MAX_PROPERTIES, R>
 where
-    T: Read + Write,
+    T: Read + ReadReady + Write,
     R: RngCore,
 {
     pub fn new(
@@ -470,10 +470,18 @@ where
             }
         }
     }
+
+    pub fn read_ready(&mut self) -> Result<bool, ReasonCode> {
+        if self.connection.is_none() {
+            return Err(ReasonCode::NetworkError);
+        }
+        let conn = self.connection.as_mut().unwrap();
+        conn.read_ready()
+    }
 }
 
 #[cfg(not(feature = "tls"))]
-async fn receive_packet<'c, T: Read + Write>(
+async fn receive_packet<'c, T: Read + ReadReady + Write>(
     buffer: &mut [u8],
     buffer_len: usize,
     recv_buffer: &mut [u8],
@@ -544,7 +552,7 @@ async fn receive_packet<'c, T: Read + Write>(
 }
 
 #[cfg(feature = "tls")]
-async fn receive_packet<'c, T: Read + Write>(
+async fn receive_packet<'c, T: Read + ReadReady + Write>(
     buffer: &mut [u8],
     buffer_len: usize,
     recv_buffer: &mut [u8],

src/network/mod.rs

@@ -23,19 +23,19 @@
  */
 
 use crate::packet::v5::reason_codes::ReasonCode;
-use embedded_io_async::{Read, Write};
+use embedded_io_async::{Read, ReadReady, Write};
 
 pub struct NetworkConnection<T>
 where
-    T: Read + Write,
+    T: Read + ReadReady + Write,
 {
     io: T,
 }
 
 /// Network connection represents an established TCP connection.
 impl<T> NetworkConnection<T>
 where
-    T: Read + Write,
+    T: Read + ReadReady + Write,
 {
     /// Create a new network handle using the provided IO implementation.
     pub fn new(io: T) -> Self {
@@ -59,4 +59,11 @@ where
             .await
             .map_err(|_| ReasonCode::NetworkError)
     }
+
+    /// Get whether a TCP connection reader is ready.
+    pub fn read_ready(&mut self) -> Result<bool, ReasonCode> {
+        self.io.read_ready()
+            .map_err(|_| ReasonCode::NetworkError)
+    }
+
 }