Integrate framing and muxer into wire protocol

radicle-node/src/client.rs

@@ -9,7 +9,7 @@ use radicle::crypto::Signer;
 use crate::clock::RefClock;
 use crate::profile::Profile;
 use crate::service::routing;
-use crate::wire::transcoder::NoHandshake;
+use crate::wire::transcode::NoHandshake;
 use crate::wire::Wire;
 use crate::{address, service};
 

radicle-node/src/wire.rs

@@ -1,5 +1,5 @@
 pub mod message;
-pub mod transcoder;
+pub mod transcode;
 
 use std::collections::{BTreeMap, HashMap, VecDeque};
 use std::convert::TryFrom;
@@ -26,7 +26,7 @@ use crate::service::{filter, routing, session};
 use crate::storage::refs::Refs;
 use crate::storage::refs::SignedRefs;
 use crate::storage::WriteStorage;
-use crate::wire::transcoder::{Handshake, HandshakeResult, Transcode};
+use crate::wire::transcode::{Framer, Handshake, HandshakeResult, MuxMsg, Transcode};
 
 /// The default type we use to represent sizes on the wire.
 ///
@@ -426,7 +426,7 @@ impl Decode for node::Features {
 
 #[derive(Debug)]
 pub struct Inbox<T: Transcode> {
-    pub transcoder: T,
+    pub pipeline: Framer<T>,
     pub deserializer: Deserializer,
 }
 
@@ -515,10 +515,11 @@ where
                         self.inner_queue
                             .push_back(nakamoto::Io::Write(*addr, reply));
                     }
+                    let pipeline = Framer::new(transcoder);
                     self.inboxes.insert(
                         *addr,
                         Inbox {
-                            transcoder,
+                            pipeline,
                             deserializer: Deserializer::new(256),
                         },
                     );
@@ -538,18 +539,31 @@ where
         }
 
         if let Some(Inbox {
-            transcoder,
+            pipeline,
             deserializer,
         }) = self.inboxes.get_mut(addr)
         {
-            let bytes = transcoder.decode(raw_bytes);
+            pipeline.input(raw_bytes);
-            deserializer.input(&bytes);
+            for frame in pipeline {
-
+                let Ok(msg) = MuxMsg::try_from(frame) else {
-            loop {
+                    // TODO: Disconnect peer.
-                match deserializer.deserialize_next() {
+                    log::error!("Message frame with invalid channel structure from {}", addr);
-                    Ok(Some(msg)) => self.inner.received_message(addr, msg),
+                    return;
-                    Ok(None) => break,
+                };
+                match msg.channel {
+                    0 => deserializer.input(&msg.data),
+                    1 => { /* TODO: Send to git worker */ }
+                    wrong_channel => {
+                        // TODO: Disconnect peer.
+                        log::error!("Wrong message channel {} from peer {}", wrong_channel, addr);
+                        return;
+                    }
+                };
+            }
 
+            for message in deserializer {
+                match message {
+                    Ok(msg) => self.inner.received_message(addr, msg),
                     Err(err) => {
                         // TODO: Disconnect peer.
                         log::error!("Invalid message received from {}: {}", addr, err);
@@ -581,11 +595,12 @@ impl<R, S, W, G, H: Handshake> Iterator for Wire<R, S, W, G, H> {
                     msg.encode(&mut buf)
                         .expect("writing to an in-memory buffer doesn't fail");
                 }
-                let Inbox { transcoder, .. } = self.inboxes.get_mut(&addr).expect(
+                let Inbox { pipeline, .. } = self.inboxes.get_mut(&addr).expect(
                     "broken handshake implementation: data sent before handshake was complete",
                 );
-                let data = transcoder.encode(buf);
+                let data = pipeline.frame(buf).expect("oversized data for a frame");
-                Some(nakamoto::Io::Write(addr, data))
+                let msg = MuxMsg { channel: 0, data };
+                Some(nakamoto::Io::Write(addr, msg.into()))
             }
             Some(Io::Event(e)) => Some(nakamoto::Io::Event(e)),
             Some(Io::Connect(a)) => Some(nakamoto::Io::Connect(a)),

radicle-node/src/wire/transcode.rs

@@ -107,8 +107,10 @@ impl<T: Transcode> Framer<T> {
         let len = u8::try_from(len).map_err(|_| OversizedData(len))?;
         let len = len.to_be_bytes();
         let mut buf = Vec::with_capacity(data.len() + 2);
+
         buf.extend(len);
         buf.append(&mut data);
+
         Ok(buf)
     }
 }
@@ -124,16 +126,20 @@ impl<T: Transcode> Iterator for Framer<T> {
         self.input
             .read_exact(&mut len)
             .expect("the length is checked");
+
         let len = u16::from_be_bytes(len) as usize;
         if self.input.len() < 2 + len {
             return None;
         }
         self.input.pop_front();
         self.input.pop_front();
+
         let reminder = self.input.split_off(len);
         let mut data = vec![0u8; len];
+
         self.input.read_exact(&mut data).expect("checked length");
         self.input = reminder;
+
         Some(data)
     }
 }
@@ -147,6 +153,17 @@ pub struct MuxMsg {
     pub data: Vec<u8>,
 }
 
+impl From<MuxMsg> for Frame {
+    fn from(mut msg: MuxMsg) -> Self {
+        let channel = msg.channel.to_be_bytes();
+        let mut data = Vec::with_capacity(msg.data.len() + 2);
+
+        data.extend(channel);
+        data.append(&mut msg.data);
+        data
+    }
+}
+
 impl TryFrom<Frame> for MuxMsg {
     type Error = ChannelError;
 
@@ -156,10 +173,13 @@ impl TryFrom<Frame> for MuxMsg {
         }
         let mut channel = [0u8; 2];
         let mut cursor = io::Cursor::new(frame);
+
         cursor
             .read_exact(&mut channel)
             .expect("the length is checked");
+
         let channel = u16::from_be_bytes(channel);
+
         Ok(MuxMsg {
             channel,
             data: cursor.into_inner(),
@@ -170,16 +190,19 @@ impl TryFrom<Frame> for MuxMsg {
 #[cfg(test)]
 mod test {
     use super::*;
+    use crate::deserializer::Deserializer;
 
     #[test]
-    fn transcode() {
+    fn decode() {
         let mut pipeline = Framer::new(PlainTranscoder);
+        let mut deser = Deserializer::<String>::new(512);
 
         let data = [
             0x00, 0x04, 0x00, 0x00, b'a', b'b', 0x00, 0x07, 0x00, 0x01, b'M', b'a', b'x', b'i',
             b'm',
         ];
         let mut expected_payloads = [(0u16, b"ab".to_vec()), (1, b"Maxim".to_vec())].into_iter();
+        let mut expected_msgs = ["ab", "Maxim"].into_iter();
 
         for byte in data {
             // Writing data byte by byte, ensuring that the reading is not broken
@@ -187,8 +210,14 @@ mod test {
             for frame in &mut pipeline {
                 let msg = MuxMsg::try_from(frame).unwrap();
                 let (channel, data) = expected_payloads.next().unwrap();
+                deser.input(&data);
                 assert_eq!(msg, MuxMsg { channel, data });
             }
         }
+
+        for msg in deser {
+            let msg = msg.unwrap();
+            assert_eq!(msg, expected_msgs.next().unwrap());
+        }
     }
 }