fix!: Rework embedded io implementations. (#86)

jettify · web-flow · commit 660aa4cec041 · 2025-12-23T21:50:10.000-05:00
diff --git a/Cargo.toml b/Cargo.toml
@@ -32,7 +32,7 @@ embedded-io-async = "0.7.0"
 [[example]]
 name = "async"
 path = "examples/local_async.rs"
-required-features = ["async"]
+required-features = ["embedded_io_async"]
 test = false
 
 
diff --git a/examples/local_async.rs b/examples/local_async.rs
@@ -4,7 +4,7 @@ use std::process::exit;
 use std::time::Duration;
 
 use tokio_serial::SerialPortBuilderExt;
-use uf_crsf::CrsfParser;
+use uf_crsf::async_io::AsyncCrsfReader;
 
 #[tokio::main]
 async fn main() {
@@ -33,12 +33,12 @@ async fn main() {
             exit(1);
         });
 
-    let mut adapted_port = FromTokio::new(&mut port);
-    let mut parser = CrsfParser::new();
+    let adapted_port = FromTokio::new(&mut port);
+    let mut reader = AsyncCrsfReader::new(adapted_port);
     println!("Reading from serial port '{}'...", path);
 
     loop {
-        match parser.read_packet(&mut adapted_port).await {
+        match reader.read_packet().await {
             Ok(packet) => {
                 println!("{:?}", packet);
             }
diff --git a/src/async_io.rs b/src/async_io.rs
@@ -1,34 +1,52 @@
 use crate::error::CrsfStreamError;
 use crate::packets::{write_packet_to_buffer, CrsfPacket, Packet, PacketAddress};
 use crate::parser::CrsfParser;
-use embedded_io_async::{Error, Read, Write};
-
-impl CrsfParser {
-    /// Asynchronously reads a complete CRSF packet from an `embedded_io_async::Read` stream.
-    ///
-    /// This function reads bytes in chunks from the provided `reader` and pushes them
-    /// into the parser one byte at time.
-    pub async fn read_packet<R: Read>(
-        &mut self,
-        reader: &mut R,
-    ) -> Result<Packet, CrsfStreamError> {
-        let mut buf = [0; 64]; // 64 is max packet size for CRSF
+use embedded_io_async::{Error, Write};
+use heapless::Deque;
+
+const ASYNC_IO_BUFFER_SIZE: usize = crate::constants::CRSF_MAX_PACKET_SIZE * 2;
+
+pub struct AsyncCrsfReader<R> {
+    parser: CrsfParser,
+    reader: R,
+    input_buffer: Deque<u8, ASYNC_IO_BUFFER_SIZE>,
+}
+
+/// Asynchronously writes a CRSF packet to an `embedded_io_async::Write` stream.
+impl<R: embedded_io_async::Read> AsyncCrsfReader<R> {
+    pub fn new(reader: R) -> Self {
+        Self {
+            parser: CrsfParser::new(),
+            reader,
+            input_buffer: Deque::new(),
+        }
+    }
+
+    pub async fn read_packet(&mut self) -> Result<Packet, CrsfStreamError> {
+        let mut temp_read_buf = [0; crate::constants::CRSF_MAX_PACKET_SIZE];
+
         loop {
-            let n = reader
-                .read(&mut buf)
+            while let Some(byte) = self.input_buffer.pop_front() {
+                match self.parser.push_byte(byte) {
+                    Ok(Some(packet)) => return Ok(packet),
+                    Ok(None) => (),
+                    Err(e) => return Err(e),
+                }
+            }
+            let bytes_read = self
+                .reader
+                .read(&mut temp_read_buf)
                 .await
                 .map_err(|e| CrsfStreamError::Io(e.kind()))?;
-            if n == 0 {
-                // This indicates a stream has closed.
+
+            if bytes_read == 0 {
                 return Err(CrsfStreamError::UnexpectedEof);
             }
 
-            for b in &buf[0..n] {
-                match self.push_byte(*b) {
-                    Ok(Some(packet)) => return Ok(packet),
-                    Ok(None) => continue,
-                    Err(e) => return Err(e),
-                }
+            for byte in &temp_read_buf[..bytes_read] {
+                self.input_buffer
+                    .push_back(*byte)
+                    .map_err(|_| CrsfStreamError::InputBufferTooSmall)?;
             }
         }
     }
diff --git a/src/blocking_io.rs b/src/blocking_io.rs
@@ -2,32 +2,49 @@ use crate::error::CrsfStreamError;
 use crate::packets::{write_packet_to_buffer, CrsfPacket, Packet, PacketAddress};
 use crate::parser::CrsfParser;
 use embedded_io::{Error, Read, Write};
+use heapless::Deque;
+
+const BLOCKING_IO_BUFFER_SIZE: usize = crate::constants::CRSF_MAX_PACKET_SIZE * 2;
+
+pub struct BlockingCrsfReader<R> {
+    parser: CrsfParser,
+    reader: R,
+    input_buffer: Deque<u8, BLOCKING_IO_BUFFER_SIZE>,
+}
+
+impl<R: Read> BlockingCrsfReader<R> {
+    pub fn new(reader: R) -> Self {
+        Self {
+            parser: CrsfParser::new(),
+            reader,
+            input_buffer: Deque::new(),
+        }
+    }
+
+    pub fn read_packet(&mut self) -> Result<Packet, CrsfStreamError> {
+        let mut temp_read_buf = [0; crate::constants::CRSF_MAX_PACKET_SIZE];
 
-impl CrsfParser {
-    /// Synchronously reads a complete CRSF packet from an `embedded_io::Read` stream.
-    ///
-    /// This function reads bytes in chunks from the provided `reader` and pushes them
-    /// into the parser one byte at time.
-    pub fn read_packet_blocking<R: Read>(
-        &mut self,
-        reader: &mut R,
-    ) -> Result<Packet, CrsfStreamError> {
-        let mut buf = [0; 64]; // 64 is max packet size for CRSF
         loop {
-            let n = reader
-                .read(&mut buf)
+            while let Some(byte) = self.input_buffer.pop_front() {
+                match self.parser.push_byte(byte) {
+                    Ok(Some(packet)) => return Ok(packet),
+                    Ok(None) => (),
+                    Err(e) => return Err(e),
+                }
+            }
+            let bytes_read = self
+                .reader
+                .read(&mut temp_read_buf)
                 .map_err(|e| CrsfStreamError::Io(e.kind()))?;
-            if n == 0 {
-                // This indicates a stream has closed.
+
+            if bytes_read == 0 {
                 return Err(CrsfStreamError::UnexpectedEof);
             }
 
-            for b in &buf[0..n] {
-                match self.push_byte(*b) {
-                    Ok(Some(packet)) => return Ok(packet),
-                    Ok(None) => continue,
-                    Err(e) => return Err(e),
-                }
+            for byte in &temp_read_buf[..bytes_read] {
+                self.input_buffer
+                    .push_back(*byte)
+                    .map_err(|_| CrsfStreamError::InputBufferTooSmall)?;
             }
         }
     }
@@ -37,7 +54,7 @@ impl CrsfParser {
 ///
 /// This function serializes the given `packet` into a temporary buffer and then
 /// writes the entire buffer to the `writer` synchronously.
-pub fn write_packet_blocking<W: Write, P: CrsfPacket>(
+pub fn write_packet<W: Write, P: CrsfPacket>(
     writer: &mut W,
     dest: PacketAddress,
     packet: &P,
diff --git a/tests/async_io_test.rs b/tests/async_io_test.rs
@@ -2,14 +2,13 @@
 #![cfg(test)]
 extern crate std;
 
-use uf_crsf::async_io::write_packet;
+use uf_crsf::async_io::{write_packet, AsyncCrsfReader};
 use uf_crsf::packets::{LinkStatistics, Packet, PacketAddress};
-use uf_crsf::parser::CrsfParser;
 use uf_crsf::CrsfStreamError;
 
-async fn build_link_statistics_packet_bytes() -> std::vec::Vec<u8> {
+async fn build_link_statistics_packet_bytes(uplink_rssi_1: u8) -> std::vec::Vec<u8> {
     let packet = LinkStatistics {
-        uplink_rssi_1: 10,
+        uplink_rssi_1: uplink_rssi_1,
         uplink_rssi_2: 20,
         uplink_link_quality: 95,
         uplink_snr: -80,
@@ -29,14 +28,14 @@ async fn build_link_statistics_packet_bytes() -> std::vec::Vec<u8> {
 
 #[tokio::test]
 async fn test_write_and_read_packet_async() {
-    let packet_bytes = build_link_statistics_packet_bytes().await;
+    let packet_bytes = build_link_statistics_packet_bytes(10).await;
 
     // Mock reader
-    let mut reader = &packet_bytes[..];
+    let reader = &packet_bytes[..];
 
     // Parser
-    let mut parser = CrsfParser::new();
-    let result = parser.read_packet(&mut reader).await;
+    let mut reader = AsyncCrsfReader::new(reader);
+    let result = reader.read_packet().await;
 
     assert!(result.is_ok());
     let parsed_packet = result.unwrap();
@@ -58,27 +57,75 @@ async fn test_write_and_read_packet_async() {
 
 #[tokio::test]
 async fn test_read_packet_async_with_no_data() {
-    let mut reader = &[][..];
-    let mut parser = CrsfParser::new();
-    let result = parser.read_packet(&mut reader).await;
+    let mut reader = AsyncCrsfReader::new(&[][..]);
+    let result = reader.read_packet().await;
     assert!(matches!(result, Err(CrsfStreamError::UnexpectedEof)));
 }
 
 #[tokio::test]
 async fn test_read_packet_async_with_incomplete_data() {
-    let packet_bytes = build_link_statistics_packet_bytes().await;
-    let mut reader = &packet_bytes[..packet_bytes.len() - 1];
-    let mut parser = CrsfParser::new();
-    let result = parser.read_packet(&mut reader).await;
+    let packet_bytes = build_link_statistics_packet_bytes(10).await;
+    let mut reader = AsyncCrsfReader::new(&packet_bytes[..packet_bytes.len() - 1]);
+    let result = reader.read_packet().await;
     assert!(matches!(result, Err(CrsfStreamError::UnexpectedEof)));
 }
 
 #[tokio::test]
 async fn test_read_packet_async_with_garbage() {
     let garbage = [0x01, 0x02, 0x03];
-    let mut reader = &garbage[..];
-    let mut parser = CrsfParser::new();
-    let result = parser.read_packet(&mut reader).await;
+    let mut reader = AsyncCrsfReader::new(&garbage[..]);
+    let result = reader.read_packet().await;
     // We expect an InvalidSync error because the first byte is not a valid sync byte.
     assert!(matches!(result, Err(CrsfStreamError::InvalidSync(_))));
 }
+
+#[tokio::test]
+async fn test_read_packet_async_chunked_stream() {
+    let packet1_bytes = build_link_statistics_packet_bytes(10).await;
+    let packet2_bytes = build_link_statistics_packet_bytes(50).await;
+
+    let mut combined_bytes = std::vec::Vec::new();
+    combined_bytes.extend_from_slice(&packet1_bytes);
+    combined_bytes.extend_from_slice(&packet2_bytes);
+
+    let mut stream_reader = AsyncCrsfReader::new(&combined_bytes[..]);
+
+    let result1 = stream_reader.read_packet().await;
+    let result2 = stream_reader.read_packet().await;
+
+    assert!(result1.is_ok());
+    let parsed_packet1 = result1.unwrap();
+    let expected_packet1 = LinkStatistics {
+        uplink_rssi_1: 10,
+        uplink_rssi_2: 20,
+        uplink_link_quality: 95,
+        uplink_snr: -80,
+        active_antenna: 1,
+        rf_mode: 2,
+        uplink_tx_power: 3,
+        downlink_rssi: 30,
+        downlink_link_quality: 98,
+        downlink_snr: -75,
+    };
+    assert!(matches!(&parsed_packet1, Packet::LinkStatistics(p) if p == &expected_packet1));
+
+    assert!(
+        result2.is_ok(),
+        "Second packet parsing failed: {:?}",
+        result2.err()
+    );
+    let parsed_packet2 = result2.unwrap();
+    let expected_packet2 = LinkStatistics {
+        uplink_rssi_1: 50,
+        uplink_rssi_2: 20,
+        uplink_link_quality: 95,
+        uplink_snr: -80,
+        active_antenna: 1,
+        rf_mode: 2,
+        uplink_tx_power: 3,
+        downlink_rssi: 30,
+        downlink_link_quality: 98,
+        downlink_snr: -75,
+    };
+    assert!(matches!(parsed_packet2, Packet::LinkStatistics(p) if p == expected_packet2));
+}
diff --git a/tests/blocking_io_test.rs b/tests/blocking_io_test.rs
@@ -2,9 +2,8 @@
 #![cfg(test)]
 extern crate std;
 
-use uf_crsf::blocking_io::write_packet_blocking;
+use uf_crsf::blocking_io::{write_packet, BlockingCrsfReader};
 use uf_crsf::packets::{LinkStatistics, Packet, PacketAddress};
-use uf_crsf::parser::CrsfParser;
 use uf_crsf::CrsfStreamError;
 
 fn build_link_statistics_packet_bytes() -> std::vec::Vec<u8> {
@@ -21,7 +20,7 @@ fn build_link_statistics_packet_bytes() -> std::vec::Vec<u8> {
         downlink_snr: -75,
     };
     let mut buffer = std::vec::Vec::new();
-    write_packet_blocking(&mut buffer, PacketAddress::FlightController, &packet).unwrap();
+    write_packet(&mut buffer, PacketAddress::FlightController, &packet).unwrap();
     buffer
 }
 
@@ -32,8 +31,8 @@ fn test_write_and_read_packet_blocking() {
     // Mock reader
     let mut reader = &packet_bytes[..];
 
-    let mut parser = CrsfParser::new();
-    let result = parser.read_packet_blocking(&mut reader);
+    let mut crsf_reader = BlockingCrsfReader::new(&mut reader);
+    let result = crsf_reader.read_packet();
 
     let parsed_packet = result.unwrap();
 
@@ -55,26 +54,26 @@ fn test_write_and_read_packet_blocking() {
 #[test]
 fn test_read_packet_blocking_with_no_data() {
     let mut reader = &[][..];
-    let mut parser = CrsfParser::new();
-    let result = parser.read_packet_blocking(&mut reader);
+    let mut crsf_reader = BlockingCrsfReader::new(&mut reader);
+    let result = crsf_reader.read_packet();
     assert!(matches!(result, Err(CrsfStreamError::UnexpectedEof)));
 }
 
 #[test]
 fn test_read_packet_blocking_with_incomplete_data() {
     let packet_bytes = build_link_statistics_packet_bytes();
     let mut reader = &packet_bytes[..packet_bytes.len() - 1];
-    let mut parser = CrsfParser::new();
-    let result = parser.read_packet_blocking(&mut reader);
+    let mut crsf_reader = BlockingCrsfReader::new(&mut reader);
+    let result = crsf_reader.read_packet();
     assert!(matches!(result, Err(CrsfStreamError::UnexpectedEof)));
 }
 
 #[test]
 fn test_read_packet_blocking_with_garbage() {
     let garbage = [0x01, 0x02, 0x03];
     let mut reader = &garbage[..];
-    let mut parser = CrsfParser::new();
-    let result = parser.read_packet_blocking(&mut reader);
+    let mut crsf_reader = BlockingCrsfReader::new(&mut reader);
+    let result = crsf_reader.read_packet();
     // We expect an InvalidSync error because the first byte is not a valid sync byte.
     assert!(matches!(result, Err(CrsfStreamError::InvalidSync(_))));
 }
