firmware: Refactor protobuf communication.

* Introduce traits in io::framed, and add Sender.

* Add io::proto module to handle protobuf-in-framed-messages.

* Add 'streaming' mode to crc module, and use it in io::proto for
  sending messages (removes need to buffer outgoing messages).

* Update dependencies.

* Simplify tasks::host_interface by using io::proto module.

firmware/Cargo.lock

@@ -245,7 +245,7 @@ dependencies = [
 [[package]]
 name = "embassy-embedded-hal"
 version = "0.3.0"
-source = "git+https://github.com/embassy-rs/embassy#94f9b2707486ca3eade5bf4b237edf3d6aa90f35"
+source = "git+https://github.com/embassy-rs/embassy#1d8e4fd970ea1794e5a726781442f13c2b2c2b66"
 dependencies = [
  "defmt 1.0.1",
  "embassy-futures",
@@ -263,7 +263,7 @@ dependencies = [
 [[package]]
 name = "embassy-executor"
 version = "0.7.0"
-source = "git+https://github.com/embassy-rs/embassy#94f9b2707486ca3eade5bf4b237edf3d6aa90f35"
+source = "git+https://github.com/embassy-rs/embassy#1d8e4fd970ea1794e5a726781442f13c2b2c2b66"
 dependencies = [
  "cortex-m",
  "critical-section",
@@ -275,7 +275,7 @@ dependencies = [
 [[package]]
 name = "embassy-executor-macros"
 version = "0.6.2"
-source = "git+https://github.com/embassy-rs/embassy#94f9b2707486ca3eade5bf4b237edf3d6aa90f35"
+source = "git+https://github.com/embassy-rs/embassy#1d8e4fd970ea1794e5a726781442f13c2b2c2b66"
 dependencies = [
  "darling",
  "proc-macro2",
@@ -286,7 +286,7 @@ dependencies = [
 [[package]]
 name = "embassy-futures"
 version = "0.1.1"
-source = "git+https://github.com/embassy-rs/embassy#94f9b2707486ca3eade5bf4b237edf3d6aa90f35"
+source = "git+https://github.com/embassy-rs/embassy#1d8e4fd970ea1794e5a726781442f13c2b2c2b66"
 dependencies = [
  "defmt 1.0.1",
 ]
@@ -294,7 +294,7 @@ dependencies = [
 [[package]]
 name = "embassy-hal-internal"
 version = "0.2.0"
-source = "git+https://github.com/embassy-rs/embassy#94f9b2707486ca3eade5bf4b237edf3d6aa90f35"
+source = "git+https://github.com/embassy-rs/embassy#1d8e4fd970ea1794e5a726781442f13c2b2c2b66"
 dependencies = [
  "cortex-m",
  "critical-section",
@@ -305,7 +305,7 @@ dependencies = [
 [[package]]
 name = "embassy-net-driver"
 version = "0.2.0"
-source = "git+https://github.com/embassy-rs/embassy#94f9b2707486ca3eade5bf4b237edf3d6aa90f35"
+source = "git+https://github.com/embassy-rs/embassy#1d8e4fd970ea1794e5a726781442f13c2b2c2b66"
 dependencies = [
  "defmt 1.0.1",
 ]
@@ -313,7 +313,7 @@ dependencies = [
 [[package]]
 name = "embassy-stm32"
 version = "0.2.0"
-source = "git+https://github.com/embassy-rs/embassy#94f9b2707486ca3eade5bf4b237edf3d6aa90f35"
+source = "git+https://github.com/embassy-rs/embassy#1d8e4fd970ea1794e5a726781442f13c2b2c2b66"
 dependencies = [
  "aligned",
  "bit_field",
@@ -361,7 +361,7 @@ dependencies = [
 [[package]]
 name = "embassy-sync"
 version = "0.7.0"
-source = "git+https://github.com/embassy-rs/embassy#94f9b2707486ca3eade5bf4b237edf3d6aa90f35"
+source = "git+https://github.com/embassy-rs/embassy#1d8e4fd970ea1794e5a726781442f13c2b2c2b66"
 dependencies = [
  "cfg-if",
  "critical-section",
@@ -375,7 +375,7 @@ dependencies = [
 [[package]]
 name = "embassy-time"
 version = "0.4.0"
-source = "git+https://github.com/embassy-rs/embassy#94f9b2707486ca3eade5bf4b237edf3d6aa90f35"
+source = "git+https://github.com/embassy-rs/embassy#1d8e4fd970ea1794e5a726781442f13c2b2c2b66"
 dependencies = [
  "cfg-if",
  "critical-section",
@@ -391,7 +391,7 @@ dependencies = [
 [[package]]
 name = "embassy-time-driver"
 version = "0.2.0"
-source = "git+https://github.com/embassy-rs/embassy#94f9b2707486ca3eade5bf4b237edf3d6aa90f35"
+source = "git+https://github.com/embassy-rs/embassy#1d8e4fd970ea1794e5a726781442f13c2b2c2b66"
 dependencies = [
  "document-features",
 ]
@@ -399,7 +399,7 @@ dependencies = [
 [[package]]
 name = "embassy-time-queue-utils"
 version = "0.1.0"
-source = "git+https://github.com/embassy-rs/embassy#94f9b2707486ca3eade5bf4b237edf3d6aa90f35"
+source = "git+https://github.com/embassy-rs/embassy#1d8e4fd970ea1794e5a726781442f13c2b2c2b66"
 dependencies = [
  "embassy-executor",
  "heapless",
@@ -408,7 +408,7 @@ dependencies = [
 [[package]]
 name = "embassy-usb-driver"
 version = "0.1.0"
-source = "git+https://github.com/embassy-rs/embassy#94f9b2707486ca3eade5bf4b237edf3d6aa90f35"
+source = "git+https://github.com/embassy-rs/embassy#1d8e4fd970ea1794e5a726781442f13c2b2c2b66"
 dependencies = [
  "defmt 1.0.1",
  "embedded-io-async",
@@ -417,7 +417,7 @@ dependencies = [
 [[package]]
 name = "embassy-usb-synopsys-otg"
 version = "0.2.0"
-source = "git+https://github.com/embassy-rs/embassy#94f9b2707486ca3eade5bf4b237edf3d6aa90f35"
+source = "git+https://github.com/embassy-rs/embassy#1d8e4fd970ea1794e5a726781442f13c2b2c2b66"
 dependencies = [
  "critical-section",
  "defmt 1.0.1",
@@ -680,6 +680,7 @@ dependencies = [
  "embedded-resources",
  "micropb",
  "micropb-gen",
+ "never",
  "panic-halt",
  "panic-probe",
  "portable-atomic",

firmware/Cargo.toml

@@ -21,6 +21,7 @@ portable-atomic = { version = "1.5", features = ["unsafe-assume-single-core"] }
 static_cell = "2.1.0"
 
 thiserror = { version = "2.0.12", default-features = false }
+never = { version = "0.1.0", default-features = false }
 micropb = { version = "0.1.1", default-features = false, features = [ "encode", "decode", "container-heapless" ] }
 
 defmt = { version = "1.0.0", optional = true }

firmware/src/crc.rs

@@ -3,17 +3,21 @@ use embassy_sync::blocking_mutex::raw::ThreadModeRawMutex;
 use embassy_sync::mutex;
 use static_cell::StaticCell;
 
-struct Shared<'a> {
-    inner: mutex::Mutex<ThreadModeRawMutex, crc::Crc<'a>>,
-}
-
-#[derive(Clone, Copy)]
-pub struct Handle {
-    shared: &'static Shared<'static>,
+struct Shared {
+    inner: mutex::Mutex<ThreadModeRawMutex, crc::Crc<'static>>,
 }
 
 static STORAGE: StaticCell<Shared> = StaticCell::new();
 
+#[derive(Clone, Copy)]
+pub struct Handle {
+    shared: &'static Shared,
+}
+
+pub struct Stream {
+    guard: mutex::MutexGuard<'static, ThreadModeRawMutex, crc::Crc<'static>>,
+}
+
 impl Handle {
     pub fn new(r: crate::CrcEngineResources) -> Self {
         let cfg = crc::Config::new(
@@ -40,4 +44,27 @@ impl Handle {
         engine.reset();
         engine.feed_bytes(data) as u16
     }
+
+    pub async fn stream(&self) -> Stream {
+        let mut guard = self.shared.inner.lock().await;
+        let engine = &mut *guard;
+
+        engine.reset();
+        Stream { guard }
+    }
+}
+
+impl Stream {
+    pub fn feed_bytes(&mut self, data: &[u8]) {
+        let engine = &mut *self.guard;
+
+        engine.feed_bytes(data);
+    }
+
+    #[allow(clippy::cast_possible_truncation)]
+    pub fn result(&self) -> u16 {
+        let engine = &*self.guard;
+
+        engine.read() as u16
+    }
 }

firmware/src/io/framed.rs

@@ -1,14 +1,34 @@
-use crate::io;
-use crate::println;
-use crate::utils;
+use crate::{io, println, utils};
 use embassy_time::Duration;
-use embedded_io::ReadReady;
+use embedded_io::{ReadReady, Write};
 use embedded_io_async::Read;
 use micropb::heapless::Vec;
 use thiserror::Error;
 
+pub trait Sender {
+    type Error;
+
+    fn send_sync(&mut self) -> Result<(), Self::Error>;
+    fn send_frame_fragment(&mut self, buf: &[u8]) -> Result<(), Self::Error>;
+}
+
+pub trait Receiver {
+    type Error;
+
+    fn buf(&self) -> &[u8];
+    fn clear_buf(&mut self);
+    fn remove_frame(&mut self, bytes: usize);
+
+    async fn receive_sync(&mut self) -> Result<(), Self::Error>;
+    async fn receive_frame_fragment(
+        &mut self,
+        fragment_pos: usize,
+        fragment_len: usize,
+    ) -> Result<(), Self::Error>;
+}
+
 #[derive(Error, Debug)]
-pub enum ReceiverError<T>
+pub enum AsyncReadReceiverError<T>
 where
     T: Read + ReadReady,
 {
@@ -20,18 +40,18 @@ where
     Other(T::Error),
 }
 
-pub struct Receiver<const N: usize, T>
+pub struct AsyncReadReceiver<const N: usize, T>
 where
     T: Read + ReadReady,
 {
-    pub buf: &'static mut Vec<u8, N>,
+    buf: &'static mut Vec<u8, N>,
     rx: T,
     sync: u8,
     first_byte_timeout: Duration,
     between_bytes_timeout: Duration,
 }
 
-impl<const N: usize, T> Receiver<N, T>
+impl<const N: usize, T> AsyncReadReceiver<N, T>
 where
     T: Read + ReadReady,
 {
@@ -50,8 +70,27 @@ where
             between_bytes_timeout,
         }
     }
+}
 
-    pub async fn wait_for_sync(&mut self) -> Result<(), ReceiverError<T>> {
+impl<const N: usize, T> Receiver for AsyncReadReceiver<N, T>
+where
+    T: Read + ReadReady,
+{
+    type Error = AsyncReadReceiverError<T>;
+
+    fn buf(&self) -> &[u8] {
+        self.buf.as_slice()
+    }
+
+    fn clear_buf(&mut self) {
+        self.buf.clear();
+    }
+
+    fn remove_frame(&mut self, bytes: usize) {
+        utils::remove_leading_bytes(self.buf, bytes);
+    }
+
+    async fn receive_sync(&mut self) -> Result<(), Self::Error> {
         // look for a sync byte in the buffer; the buffer may have
         // leftover bytes from a previous cycle (due to an incomplete
         // message, or a message which failed validation)
@@ -77,7 +116,7 @@ where
                     }
                     Err(e) => {
                         println!("read error");
-                        return Err(ReceiverError::Other(e));
+                        return Err(AsyncReadReceiverError::Other(e));
                     }
                 }
             }
@@ -85,11 +124,11 @@ where
         Ok(())
     }
 
-    pub async fn read_frame_fragment(
+    async fn receive_frame_fragment(
         &mut self,
         fragment_pos: usize,
         fragment_len: usize,
-    ) -> Result<(), ReceiverError<T>> {
+    ) -> Result<(), Self::Error> {
         // compute the number of bytes available in the buffer, and the
         // number of bytes to read
         let bytes_available = self.buf[fragment_pos..].len();
@@ -102,7 +141,7 @@ where
         // if the fragment will not fit in the buffer's available space,
         // return an error
         if bytes_to_read > (self.buf.capacity() - self.buf.len()) {
-            return Err(ReceiverError::BufferCapacity);
+            return Err(AsyncReadReceiverError::BufferCapacity);
         }
 
         // wait for the needed bytes to arrive, with timeouts
@@ -127,7 +166,7 @@ where
                 unsafe {
                     self.buf.set_len(read_start + bytes_read);
                 }
-                Err(ReceiverError::Timeout)
+                Err(AsyncReadReceiverError::Timeout)
             }
             Err(io::ReadError::Other { bytes_read, source }) => {
                 // set the buffer length to include the number of bytes
@@ -135,8 +174,41 @@ where
                 unsafe {
                     self.buf.set_len(read_start + bytes_read);
                 }
-                Err(ReceiverError::Other(source))
+                Err(AsyncReadReceiverError::Other(source))
             }
         }
     }
 }
+
+pub struct WriteSender<T>
+where
+    T: Write,
+{
+    tx: T,
+    sync: u8,
+}
+
+impl<T> WriteSender<T>
+where
+    T: Write,
+{
+    pub fn new(tx: T, sync: u8) -> Self {
+        Self { tx, sync }
+    }
+}
+
+impl<T> Sender for WriteSender<T>
+where
+    T: Write,
+{
+    type Error = T::Error;
+
+    fn send_sync(&mut self) -> Result<(), Self::Error> {
+        let buf = [self.sync; 1];
+        self.tx.write_all(&buf)
+    }
+
+    fn send_frame_fragment(&mut self, buf: &[u8]) -> Result<(), Self::Error> {
+        self.tx.write_all(buf)
+    }
+}

firmware/src/io/mod.rs

@@ -5,6 +5,7 @@ use embedded_io_async::Read;
 use thiserror::Error;
 
 pub mod framed;
+pub mod proto;
 
 #[derive(Error, Debug)]
 pub enum ReadError<T>

firmware/src/io/proto.rs

@@ -0,0 +1,128 @@
+use byteorder::{ByteOrder, LittleEndian};
+use micropb::{DecodeError, MessageDecode, MessageEncode, PbDecoder, PbEncoder, PbWrite};
+use never;
+use thiserror::Error;
+
+use super::framed;
+use crate::{crc, proto};
+
+pub const SYNC_BYTE: u8 = 0xfc;
+pub const MESSAGE_LENGTH_SIZE: usize = 2;
+pub const MESSAGE_CRC_SIZE: usize = 2;
+
+struct Writer<'a, T>
+where
+    T: framed::Sender,
+{
+    ftx: &'a mut T,
+    crc: &'a mut crc::Stream,
+}
+
+impl<T> PbWrite for Writer<'_, T>
+where
+    T: framed::Sender,
+{
+    type Error = T::Error;
+
+    fn pb_write(&mut self, data: &[u8]) -> Result<(), T::Error> {
+        self.ftx.send_frame_fragment(data)?;
+        self.crc.feed_bytes(data);
+        Ok(())
+    }
+}
+
+pub async fn send_target_message<T>(
+    ftx: &mut T,
+    msg: proto::api_::TargetMessage,
+    crc: crc::Handle,
+) -> Result<(), T::Error>
+where
+    T: framed::Sender,
+{
+    ftx.send_sync()?;
+
+    #[allow(clippy::cast_possible_truncation)]
+    let encoded_size = (msg.compute_size() as u16).to_le_bytes();
+    ftx.send_frame_fragment(&encoded_size)?;
+
+    let mut crc_stream = crc.stream().await;
+    let mut writer = Writer {
+        ftx,
+        crc: &mut crc_stream,
+    };
+    let mut encoder = PbEncoder::new(&mut writer);
+    msg.encode(&mut encoder)?;
+
+    let crc_result = crc_stream.result().to_le_bytes();
+    ftx.send_frame_fragment(&crc_result)
+}
+
+#[derive(Error, Debug)]
+pub enum ReceiveError<T>
+where
+    T: framed::Receiver,
+{
+    #[error("sync error")]
+    Sync,
+    #[error("CRC-16 mismatch")]
+    Crc,
+    #[error("framing error")]
+    Framing(T::Error),
+    #[error("decode error")]
+    Decode(DecodeError<never::Never>),
+}
+
+pub async fn receive_host_message<T>(
+    frx: &mut T,
+    crc: crc::Handle,
+) -> Result<proto::api_::HostMessage, ReceiveError<T>>
+where
+    T: framed::Receiver,
+{
+    frx.receive_sync().await.or(Err(ReceiveError::Sync))?;
+
+    // sync byte was seen, read the length of the message
+    frx.receive_frame_fragment(0, MESSAGE_LENGTH_SIZE)
+        .await
+        .map_err(ReceiveError::Framing)?;
+
+    // get the message length from the buffer
+    let message_len = LittleEndian::read_u16(&frx.buf()[..MESSAGE_LENGTH_SIZE]) as usize;
+
+    // read the message
+    let message_pos = MESSAGE_LENGTH_SIZE;
+    frx.receive_frame_fragment(message_pos, message_len)
+        .await
+        .map_err(ReceiveError::Framing)?;
+
+    // read the CRC-16 of the message
+    let crc_pos = message_pos + message_len;
+    frx.receive_frame_fragment(crc_pos, MESSAGE_CRC_SIZE)
+        .await
+        .map_err(ReceiveError::Framing)?;
+
+    // get the expected CRC-16 from the buffer
+    let expected_crc = LittleEndian::read_u16(&frx.buf()[crc_pos..crc_pos + MESSAGE_CRC_SIZE]);
+    // compute the actual CRC-16
+    let computed_crc = crc
+        .compute(&frx.buf()[message_pos..message_pos + message_len])
+        .await;
+    // check the CRC-16
+    if computed_crc != expected_crc {
+        return Err(ReceiveError::Crc);
+    }
+
+    // at this point the buffer has been confirmed to contain a
+    // valid message frame, so any failures beyond this point can
+    // only drop the message, it cannot be re-parsed
+    let mut decoder = PbDecoder::new(&frx.buf()[message_pos..message_pos + message_len]);
+    let mut msg = proto::api_::HostMessage::default();
+    let result = msg
+        .decode(&mut decoder, message_len)
+        .map_err(ReceiveError::Decode)
+        .map(|_x| msg);
+
+    frx.remove_frame(crc_pos + MESSAGE_CRC_SIZE);
+
+    result
+}

firmware/src/tasks/host_interface/mod.rs

@@ -1,20 +1,17 @@
-use crate::{crc, io, println, utils};
-use byteorder::{ByteOrder, LittleEndian};
+use crate::io::framed::Receiver;
+use crate::{crc, io, println};
 use embassy_executor::Spawner;
 use embassy_stm32::{gpio, usart};
 use embassy_time::Duration;
-use embedded_io_async::Write;
 use static_cell::StaticCell;
 
-use micropb::{DecodeError, MessageDecode, MessageEncode, PbDecoder, PbEncoder, heapless::Vec};
+use micropb::{DecodeError, heapless::Vec};
 
 use crate::proto::api_::*;
 use crate::proto::echo_::*;
 
-const SYNC_BYTE: u8 = 0xfc;
-const MESSAGE_LENGTH_SIZE: usize = 2;
-const MESSAGE_CRC_SIZE: usize = 2;
-const MAX_INBOUND_MESSAGE_SIZE: usize = MESSAGE_LENGTH_SIZE + MESSAGE_CRC_SIZE + 2048;
+const MAX_INBOUND_MESSAGE_SIZE: usize =
+    io::proto::MESSAGE_LENGTH_SIZE + io::proto::MESSAGE_CRC_SIZE + 2048;
 
 struct UartBuf {
     buffer: [u8; 128],
@@ -100,201 +97,89 @@ pub fn start(
 }
 
 #[embassy_executor::task]
-async fn tx_task(mut tx: Tx) {
-    println!("tx_task started");
-    let mut buf = Vec::<u8, 128>::new();
+async fn tx_task(tx: Tx) {
+    let mut ftx = io::framed::WriteSender::new(tx.uart, io::proto::SYNC_BYTE);
 
     loop {
         let msg = tx.channel.receive().await;
 
-        #[allow(clippy::cast_possible_truncation)]
-        let encoded_size = (msg.compute_size() as u16).to_le_bytes();
-
-        buf.clear();
-
-        unsafe {
-            // sync byte
-            buf.push_unchecked(SYNC_BYTE);
-            // two bytes for length of message (not including CRC)
-            buf.push_unchecked(encoded_size[0]);
-            buf.push_unchecked(encoded_size[1]);
-        }
-
-        let mut encoder = PbEncoder::new(&mut buf);
-        msg.encode(&mut encoder).unwrap();
-
-        let crc = (tx.crc.compute(&buf[3..]).await).to_le_bytes();
-        buf.push(crc[0]).unwrap();
-        buf.push(crc[1]).unwrap();
-
-        tx.uart.write_all(&buf).await.unwrap();
+        let _ = io::proto::send_target_message(&mut ftx, msg, tx.crc).await;
     }
 }
 
 #[embassy_executor::task]
-#[allow(clippy::too_many_lines)]
 async fn rx_task(rx: Rx) {
-    println!("rx_task started");
-
-    let mut frx = io::framed::Receiver::new(
+    let mut frx = io::framed::AsyncReadReceiver::new(
         rx.message_buf,
         rx.uart,
-        SYNC_BYTE,
+        io::proto::SYNC_BYTE,
         Duration::from_millis(250),
         Duration::from_millis(50),
     );
 
     loop {
-        if frx.wait_for_sync().await.is_err() {
-            println!("error while waiting for sync");
-            continue;
-        }
-        println!("received sync byte from host");
-
-        // sync byte was seen, read the length of the message; if
-        // there aren't enough bytes in the buffer, wait for them,
-        // with timeouts
-        match frx.read_frame_fragment(0, MESSAGE_LENGTH_SIZE).await {
-            Ok(()) => {}
-            Err(io::framed::ReceiverError::BufferCapacity) => {
-                println!("insufficient space to read message length");
-                frx.buf.clear();
-                continue;
-            }
-            Err(io::framed::ReceiverError::Timeout) => {
-                // restart loop, since the sync byte seen wasn't
-                // the beginning of a message, or the sender
-                // stopped sending
-                println!("timeout waiting for message length");
-                continue;
-            }
-            Err(io::framed::ReceiverError::Other(_)) => {
-                // restart loop when a USART error occurs
-                println!("usart error waiting for message length");
-                continue;
-            }
-        }
-
-        // get the message length from the buffer
-        let message_len = LittleEndian::read_u16(&frx.buf[..MESSAGE_LENGTH_SIZE]) as usize;
-        println!("receiving message of {=u16} bytes", message_len as u16);
-
-        // read the message; if there aren't enough bytes in the
-        // buffer for the message, wait for them, with timeouts
-        let message_pos = MESSAGE_LENGTH_SIZE;
-        match frx.read_frame_fragment(message_pos, message_len).await {
-            Ok(()) => {}
-            Err(io::framed::ReceiverError::BufferCapacity) => {
-                println!("insufficient space to read message body");
-                frx.buf.clear();
-                continue;
-            }
-            Err(io::framed::ReceiverError::Timeout) => {
-                // restart loop, since the sync byte seen wasn't
-                // the beginning of a message, or the sender
-                // stopped sending
-                println!("timeout waiting for message body");
-                continue;
-            }
-            Err(io::framed::ReceiverError::Other(_)) => {
-                // restart loop when a USART error occurs
-                println!("usart error waiting for message length");
-                continue;
-            }
-        }
-
-        // read the CRC-16 of the message; if there aren't enough
-        // bytes in the buffer, wait for them, with timeouts
-        let crc_pos = message_pos + message_len;
-        match frx.read_frame_fragment(crc_pos, MESSAGE_CRC_SIZE).await {
-            Ok(()) => {}
-            Err(io::framed::ReceiverError::BufferCapacity) => {
-                println!("insufficient space to read CRC-16");
-                frx.buf.clear();
-                continue;
-            }
-            Err(io::framed::ReceiverError::Timeout) => {
-                // restart loop, since the sync byte seen wasn't
-                // the beginning of a message, or the sender
-                // stopped sending
-                println!("timeout waiting for message CRC-16");
-                continue;
-            }
-            Err(io::framed::ReceiverError::Other(_)) => {
-                // restart loop when a USART error occurs
-                println!("usart error waiting for message length");
-                continue;
-            }
-        }
-
-        // get the expected CRC-16 from the buffer
-        let expected_crc = LittleEndian::read_u16(&frx.buf[crc_pos..crc_pos + MESSAGE_CRC_SIZE]);
-        // compute the actual CRC-16
-        let computed_crc = rx
-            .crc
-            .compute(&frx.buf[message_pos..message_pos + message_len])
-            .await;
-
-        if computed_crc != expected_crc {
-            println!(
-                "CRC-16 mismatch: expected {=u16:#x} - computed {=u16:#x}",
-                expected_crc, computed_crc
-            );
-            continue;
-        }
-
-        println!("message frame is valid");
-
-        // at this point the buffer has been confirmed to contain a
-        // valid message frame, so any failures beyond this point can
-        // only drop the message, it cannot be re-parsed
-        let mut decoder = PbDecoder::new(&frx.buf[message_pos..message_pos + message_len]);
-        let mut msg = HostMessage::default();
-        match msg.decode(&mut decoder, message_len) {
-            Ok(()) => {
-                println!("message decoded");
+        match io::proto::receive_host_message(&mut frx, rx.crc).await {
+            Ok(msg) => {
                 handle_host_message(msg, rx.channel).await;
             }
-            Err(DecodeError::VarIntLimit) => {
-                println!("decoder error: VarInt too long");
-            }
-            Err(DecodeError::UnexpectedEof) => {
-                println!("decoder error: unexpected EOF");
-            }
-            Err(DecodeError::Deprecation) => {
-                println!("decoder error: deprecated wire type found");
-            }
-            Err(DecodeError::UnknownWireType) => {
-                println!("decoder error: unknown wire type found");
-            }
-            Err(DecodeError::ZeroField) => {
-                println!("decoder error: invalid field number 0");
-            }
-            Err(DecodeError::Utf8) => {
-                println!("decoder error: invalid UTF-8 string");
-            }
-            Err(DecodeError::Capacity) => {
-                println!("decoder error: capacity of fixed-length field exceeed");
-            }
-            Err(DecodeError::WrongLen) => {
-                println!("decoder error: length-delimited-record too long");
-            }
-            Err(_) => {
-                println!("decoder error: unknown");
+            Err(io::proto::ReceiveError::Sync | io::proto::ReceiveError::Crc) => {}
+            Err(io::proto::ReceiveError::Framing(e)) => {
+                match e {
+                    io::framed::AsyncReadReceiverError::BufferCapacity => {
+                        // the host message will not fit in the
+                        // buffer, so clear the buffer and restart the
+                        // loop
+                        frx.clear_buf();
+                    }
+                    #[allow(clippy::match_same_arms)]
+                    io::framed::AsyncReadReceiverError::Timeout => {
+                        // restart loop, since the sync byte seen wasn't
+                        // the beginning of a message, or the sender
+                        // stopped sending
+                    }
+                    #[allow(clippy::match_same_arms)]
+                    io::framed::AsyncReadReceiverError::Other(_) => {
+                        // restart loop when a USART error occurs
+                    }
+                }
             }
+            Err(io::proto::ReceiveError::Decode(e)) => match e {
+                DecodeError::VarIntLimit => {
+                    println!("decoder error: VarInt too long");
+                }
+                DecodeError::UnexpectedEof => {
+                    println!("decoder error: unexpected EOF");
+                }
+                DecodeError::Deprecation => {
+                    println!("decoder error: deprecated wire type found");
+                }
+                DecodeError::UnknownWireType => {
+                    println!("decoder error: unknown wire type found");
+                }
+                DecodeError::ZeroField => {
+                    println!("decoder error: invalid field number 0");
+                }
+                DecodeError::Utf8 => {
+                    println!("decoder error: invalid UTF-8 string");
+                }
+                DecodeError::Capacity => {
+                    println!("decoder error: capacity of fixed-length field exceeed");
+                }
+                DecodeError::WrongLen => {
+                    println!("decoder error: length-delimited-record too long");
+                }
+                _ => {
+                    println!("decoder error: unknown");
+                }
+            },
         }
-
-        utils::remove_leading_bytes(frx.buf, crc_pos + MESSAGE_CRC_SIZE);
     }
 }
 
 async fn handle_host_message(host: HostMessage, generator: crate::TargetMessageGenerator) {
     match host.msg {
-        None => {
-            println!("message is empty, ignoring");
-        }
+        None => {}
         Some(HostMessage_::Msg::Echo(echo)) => {
-            println!("replying to echo request with data: {=u32:#x}", echo.data);
             generator
                 .send(TargetMessage {
                     msg: Some(TargetMessage_::Msg::Echo(EchoResponse { data: echo.data })),