rch::io: use proper codec

surban · surban · commit 7b00cc7dfd32 · 2026-01-16T21:23:02.000+01:00
diff --git a/remoc/src/rch/io/mod.rs b/remoc/src/rch/io/mod.rs
@@ -125,6 +125,7 @@
 use serde::{Deserialize, Serialize};
 
 use super::{bin, oneshot};
+use crate::codec;
 
 mod receiver;
 mod sender;
@@ -136,11 +137,13 @@ use sender::SizeMode;
 
 /// Internal enum to track size information on the receiver side.
 #[derive(Debug, Serialize, Deserialize)]
-pub(super) enum SizeInfo {
+#[serde(bound(serialize = "Codec: codec::Codec"))]
+#[serde(bound(deserialize = "Codec: codec::Codec"))]
+pub(super) enum SizeInfo<Codec> {
     /// Size was known at channel creation.
     Determined(u64),
     /// Size will be received when sender shuts down.
-    Undetermined(oneshot::Receiver<u64, crate::codec::Default>),
+    Undetermined(oneshot::Receiver<u64, Codec>),
 }
 
 /// Creates a new I/O channel with unknown size.
@@ -150,7 +153,10 @@ pub(super) enum SizeInfo {
 /// The receiver cannot know the size in advance (returns `None` from [`Receiver::size`]).
 ///
 /// Both ends can be sent to remote endpoints.
-pub fn channel() -> (Sender, Receiver) {
+pub fn channel<Codec>() -> (Sender<Codec>, Receiver<Codec>)
+where
+    Codec: codec::Codec,
+{
     let (bin_tx, bin_rx) = bin::channel();
     let (size_tx, size_rx) = oneshot::channel();
 
@@ -171,7 +177,10 @@ pub fn channel() -> (Sender, Receiver) {
 /// The receiver can query the size via [`Receiver::size`], which returns `Some(size)`.
 ///
 /// Both ends can be sent to remote endpoints.
-pub fn sized(size: u64) -> (Sender, Receiver) {
+pub fn sized<Codec>(size: u64) -> (Sender<Codec>, Receiver<Codec>)
+where
+    Codec: codec::Codec,
+{
     let (bin_tx, bin_rx) = bin::channel();
 
     let sender = Sender::new(bin_tx, SizeMode::Known(size));
diff --git a/remoc/src/rch/io/receiver.rs b/remoc/src/rch/io/receiver.rs
@@ -11,17 +11,17 @@ use tokio::io::AsyncRead;
 use tokio_util::sync::ReusableBoxFuture;
 
 use super::{SizeInfo, bin, oneshot};
-use crate::chmux::DataBuf;
+use crate::{chmux::DataBuf, codec};
 
 /// An I/O channel receiver that implements [`AsyncRead`].
 ///
 /// Reads binary data from the underlying channel.
 /// Tracks bytes read and verifies size on completion.
-pub struct Receiver {
+pub struct Receiver<Codec = codec::Default> {
     /// The underlying binary channel receiver. Uses Mutex for serialization with &self.
     bin_receiver: Mutex<Option<bin::Receiver>>,
     /// Size information (known or to be received). Uses Option to allow taking during EOF handling.
-    size_info: Mutex<Option<SizeInfo>>,
+    size_info: Mutex<Option<SizeInfo<Codec>>>,
     /// Total bytes read so far.
     bytes_read: u64,
     /// Current data buffer being consumed.
@@ -38,7 +38,7 @@ enum ReceiverState {
     VerifyingSize(ReusableBoxFuture<'static, Result<u64, io::Error>>),
 }
 
-impl fmt::Debug for Receiver {
+impl<Codec> fmt::Debug for Receiver<Codec> {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
         f.debug_struct("Receiver")
             .field("size", &self.size())
@@ -50,16 +50,18 @@ impl fmt::Debug for Receiver {
 
 /// A receiver in transport.
 #[derive(Debug, Serialize, Deserialize)]
-pub(crate) struct TransportedReceiver {
+#[serde(bound(serialize = "Codec: codec::Codec"))]
+#[serde(bound(deserialize = "Codec: codec::Codec"))]
+pub(crate) struct TransportedReceiver<Codec> {
     /// The underlying binary receiver.
     bin_receiver: bin::Receiver,
     /// Size info for transport.
-    size: SizeInfo,
+    size: SizeInfo<Codec>,
 }
 
-impl Receiver {
+impl<Codec> Receiver<Codec> {
     /// Creates a new receiver.
-    pub(super) fn new(bin_receiver: bin::Receiver, size_info: SizeInfo) -> Self {
+    pub(super) fn new(bin_receiver: bin::Receiver, size_info: SizeInfo<Codec>) -> Self {
         Self {
             bin_receiver: Mutex::new(Some(bin_receiver)),
             size_info: Mutex::new(Some(size_info)),
@@ -104,11 +106,14 @@ async fn receive_data(mut bin_receiver: bin::Receiver) -> Result<(Option<DataBuf
     Ok((data, bin_receiver))
 }
 
-async fn receive_size(size_rx: oneshot::Receiver<u64, crate::codec::Default>) -> Result<u64, io::Error> {
+async fn receive_size<Codec: codec::Codec>(size_rx: oneshot::Receiver<u64, Codec>) -> Result<u64, io::Error> {
     size_rx.await.map_err(|e| io::Error::new(ErrorKind::UnexpectedEof, e.to_string()))
 }
 
-impl Receiver {
+impl<Codec> Receiver<Codec>
+where
+    Codec: codec::Codec,
+{
     /// Polls to complete any pending receive or verify operation.
     fn poll_complete(&mut self, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
         if let ReceiverState::Receiving(ref mut fut) = self.state {
@@ -175,7 +180,10 @@ impl Receiver {
     }
 }
 
-impl AsyncRead for Receiver {
+impl<Codec> AsyncRead for Receiver<Codec>
+where
+    Codec: codec::Codec,
+{
     fn poll_read(
         mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut tokio::io::ReadBuf<'_>,
     ) -> Poll<io::Result<()>> {
@@ -232,7 +240,10 @@ impl AsyncRead for Receiver {
     }
 }
 
-impl Serialize for Receiver {
+impl<Codec> Serialize for Receiver<Codec>
+where
+    Codec: codec::Codec,
+{
     fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
     where
         S: serde::Serializer,
@@ -249,16 +260,19 @@ impl Serialize for Receiver {
             .take()
             .ok_or_else(|| serde::ser::Error::custom("cannot serialize: size info already consumed"))?;
 
-        TransportedReceiver { bin_receiver, size }.serialize(serializer)
+        TransportedReceiver::<Codec> { bin_receiver, size }.serialize(serializer)
     }
 }
 
-impl<'de> Deserialize<'de> for Receiver {
+impl<'de, Codec> Deserialize<'de> for Receiver<Codec>
+where
+    Codec: codec::Codec,
+{
     fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
     where
         D: serde::Deserializer<'de>,
     {
-        let transported = TransportedReceiver::deserialize(deserializer)?;
+        let transported = TransportedReceiver::<Codec>::deserialize(deserializer)?;
         Ok(Self::new(transported.bin_receiver, transported.size))
     }
 }
diff --git a/remoc/src/rch/io/sender.rs b/remoc/src/rch/io/sender.rs
@@ -12,26 +12,29 @@ use tokio::io::AsyncWrite;
 use tokio_util::sync::ReusableBoxFuture;
 
 use super::{bin, oneshot};
+use crate::codec;
 
 /// Size handling mode for the sender.
 #[derive(Debug, Serialize, Deserialize)]
-pub(super) enum SizeMode {
+#[serde(bound(serialize = "Codec: codec::Codec"))]
+#[serde(bound(deserialize = "Codec: codec::Codec"))]
+pub(super) enum SizeMode<Codec> {
     /// Size is known (either upfront or after shutdown).
     Known(u64),
     /// Size is unknown. Will send final byte count via oneshot on shutdown,
     /// then transition to Known.
-    Unknown(oneshot::Sender<u64, crate::codec::Default>),
+    Unknown(oneshot::Sender<u64, Codec>),
 }
 
 /// An I/O channel sender that implements [`AsyncWrite`].
 ///
 /// Writes binary data to the underlying channel.
 /// Tracks bytes written and enforces size limits if specified.
-pub struct Sender {
+pub struct Sender<Codec = codec::Default> {
     /// The underlying binary channel sender. Uses Mutex for serialization with &self.
     bin_sender: Mutex<Option<bin::Sender>>,
     /// Size handling mode.
-    size_mode: Mutex<SizeMode>,
+    size_mode: Mutex<SizeMode<Codec>>,
     /// Total bytes written so far.
     bytes_written: u64,
     /// Cached chunk size from chmux sender.
@@ -42,7 +45,7 @@ pub struct Sender {
     sending: Option<ReusableBoxFuture<'static, Result<(bin::Sender, u64), io::Error>>>,
 }
 
-impl fmt::Debug for Sender {
+impl<Codec> fmt::Debug for Sender<Codec> {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
         f.debug_struct("Sender")
             .field("expected_size", &self.expected_size())
@@ -53,18 +56,20 @@ impl fmt::Debug for Sender {
 
 /// A sender in transport.
 #[derive(Debug, Serialize, Deserialize)]
-pub(crate) struct TransportedSender {
+#[serde(bound(serialize = "Codec: codec::Codec"))]
+#[serde(bound(deserialize = "Codec: codec::Codec"))]
+pub(crate) struct TransportedSender<Codec> {
     /// The underlying binary sender. None if already closed.
     bin_sender: Option<bin::Sender>,
     /// Size handling mode.
-    size_mode: SizeMode,
+    size_mode: SizeMode<Codec>,
     /// Total bytes written so far.
     bytes_written: u64,
 }
 
-impl Sender {
+impl<Codec> Sender<Codec> {
     /// Creates a new sender.
-    pub(super) fn new(bin_sender: bin::Sender, size_mode: SizeMode) -> Self {
+    pub(super) fn new(bin_sender: bin::Sender, size_mode: SizeMode<Codec>) -> Self {
         Self {
             bin_sender: Mutex::new(Some(bin_sender)),
             size_mode: Mutex::new(size_mode),
@@ -109,7 +114,7 @@ async fn connect_sender(mut bin_sender: bin::Sender) -> Result<(bin::Sender, usi
     Ok((bin_sender, chunk_size))
 }
 
-impl Sender {
+impl<Codec> Sender<Codec> {
     /// Polls to complete any pending connect or send operations.
     fn poll_complete(&mut self, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
         if let Some(future) = &mut self.connecting {
@@ -154,7 +159,10 @@ impl Sender {
     }
 }
 
-impl AsyncWrite for Sender {
+impl<Codec> AsyncWrite for Sender<Codec>
+where
+    Codec: codec::Codec,
+{
     fn poll_write(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8]) -> Poll<io::Result<usize>> {
         let this = self.as_mut().get_mut();
 
@@ -234,7 +242,10 @@ impl AsyncWrite for Sender {
     }
 }
 
-impl Serialize for Sender {
+impl<Codec> Serialize for Sender<Codec>
+where
+    Codec: codec::Codec,
+{
     fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
     where
         S: serde::Serializer,
@@ -245,16 +256,20 @@ impl Serialize for Sender {
             SizeMode::Known(0), // Placeholder, sender is consumed anyway
         );
 
-        TransportedSender { bin_sender, size_mode, bytes_written: self.bytes_written }.serialize(serializer)
+        TransportedSender::<Codec> { bin_sender, size_mode, bytes_written: self.bytes_written }
+            .serialize(serializer)
     }
 }
 
-impl<'de> Deserialize<'de> for Sender {
+impl<'de, Codec> Deserialize<'de> for Sender<Codec>
+where
+    Codec: codec::Codec,
+{
     fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
     where
         D: serde::Deserializer<'de>,
     {
-        let transported = TransportedSender::deserialize(deserializer)?;
+        let transported = TransportedSender::<Codec>::deserialize(deserializer)?;
 
         Ok(Self {
             bin_sender: Mutex::new(transported.bin_sender),
