feat: don't split header and body across TCP packets

yamux/src/connection.rs

@@ -500,7 +500,7 @@ impl<T: AsyncRead + AsyncWrite + Unpin> Active<T> {
         }
     }
 
-    fn poll_new_outbound(&mut self, cx: &mut Context<'_>) -> Poll<Result<Stream>> {
+    fn poll_new_outbound(&mut self, cx: &Context<'_>) -> Poll<Result<Stream>> {
         if self.streams.len() >= self.config.max_num_streams {
             log::error!("{}: maximum number of streams reached", self.id);
             return Poll::Ready(Err(ConnectionError::TooManyStreams));
@@ -722,7 +722,7 @@ impl<T: AsyncRead + AsyncWrite + Unpin> Active<T> {
                     shared.update_state(self.id, stream_id, State::RecvClosed);
                 }
                 shared.window = shared.window.saturating_sub(frame.body_len());
-                shared.buffer.push(frame.into_body());
+                shared.buffer.push(frame.into_body().into());
 
                 if matches!(self.config.window_update_mode, WindowUpdateMode::OnReceive) {
                     if let Some(credit) = shared.next_window_update() {
@@ -765,7 +765,7 @@ impl<T: AsyncRead + AsyncWrite + Unpin> Active<T> {
                 return Action::Reset(Frame::new(header));
             }
             shared.window = shared.window.saturating_sub(frame.body_len());
-            shared.buffer.push(frame.into_body());
+            shared.buffer.push(frame.into_body().into());
             if let Some(w) = shared.reader.take() {
                 w.wake()
             }

yamux/src/connection/cleanup.rs

@@ -33,7 +33,7 @@ impl Future for Cleanup {
     type Output = ConnectionError;
 
     fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        let mut this = self.get_mut();
+        let this = self.get_mut();
 
         loop {
             match this.state {

yamux/src/connection/closing.rs

@@ -45,7 +45,7 @@ where
     type Output = Result<()>;
 
     fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        let mut this = self.get_mut();
+        let this = self.get_mut();
 
         loop {
             match this.state {

yamux/src/frame.rs

@@ -12,11 +12,12 @@ pub mod header;
 mod io;
 
 use futures::future::Either;
-use header::{Data, GoAway, Header, Ping, StreamId, WindowUpdate};
+use header::{Data, GoAway, Header, StreamId, WindowUpdate};
 use std::{convert::TryInto, num::TryFromIntError};
 
 pub use io::FrameDecodeError;
 pub(crate) use io::Io;
+use crate::frame::header::{HEADER_SIZE, Ping};
 
 /// A Yamux message frame consisting of header and body.
 #[derive(Clone, Debug, PartialEq, Eq)]
@@ -32,7 +33,6 @@ impl<T> Frame<T> {
             body: Vec::new(),
         }
     }
-
     pub fn header(&self) -> &Header<T> {
         &self.header
     }
@@ -56,6 +56,14 @@ impl<T> Frame<T> {
             body: self.body,
         }
     }
+
+    pub(crate) fn body(&self) -> &[u8] {
+        self.body.as_slice()
+    }
+
+    pub(crate) fn len(&self) -> usize {
+        self.body.len() + HEADER_SIZE
+    }
 }
 
 impl<A: header::private::Sealed> From<Frame<A>> for Frame<()> {
@@ -68,6 +76,23 @@ impl<A: header::private::Sealed> From<Frame<A>> for Frame<()> {
 }
 
 impl Frame<()> {
+    pub(crate) fn try_from_header_buffer(
+        buffer: &[u8; HEADER_SIZE],
+        max_body_len: usize,
+    ) -> Result<Either<Frame<()>, Frame<Data>>, FrameDecodeError> {
+        let header = header::decode(buffer)?;
+
+        let either = match header.try_into_data() {
+            Ok(data) if data.body_len() > max_body_len => {
+                return Err(FrameDecodeError::FrameTooLarge(data.body_len()));
+            }
+            Ok(data) => Either::Right(Frame::new(data)),
+            Err(other) => Either::Left(Frame::new(other)),
+        };
+
+        Ok(either)
+    }
+
     pub(crate) fn into_data(self) -> Frame<Data> {
         Frame {
             header: self.header.into_data(),
@@ -108,14 +133,17 @@ impl Frame<Data> {
         Frame::new(header)
     }
 
-    pub fn body(&self) -> &[u8] {
-        &self.body
+    pub fn body_mut(&mut self) -> &mut [u8] {
+        self.body.as_mut_slice()
     }
 
     pub fn body_len(&self) -> u32 {
-        // Safe cast since we construct `Frame::<Data>`s only with
-        // `Vec<u8>` of length [0, u32::MAX] in `Frame::data` above.
-        self.body().len() as u32
+        let len_in_header = self.header.body_len();
+        let actual_len = self.body.len();
+
+        // debug_assert_eq!(len_in_header, actual_len);
+
+        len_in_header as u32
     }
 
     pub fn into_body(self) -> Vec<u8> {

yamux/src/frame/header.rs

@@ -48,13 +48,9 @@ impl<T> Header<T> {
         self.stream_id
     }
 
-    pub fn len(&self) -> Len {
-        self.length
-    }
-
     #[cfg(test)]
-    pub fn set_len(&mut self, len: u32) {
-        self.length = Len(len)
+    pub fn set_len(&mut self, len: usize) {
+        self.length = Len(len as u32)
     }
 
     /// Arbitrary type cast, use with caution.
@@ -92,6 +88,14 @@ impl Header<()> {
         self.cast()
     }
 
+    pub(crate) fn try_into_data(self) -> Result<Header<Data>, Self> {
+        if self.tag == Tag::Data {
+            return Ok(self.into_data());
+        }
+
+        Err(self)
+    }
+
     pub(crate) fn into_window_update(self) -> Header<WindowUpdate> {
         debug_assert_eq!(self.tag, Tag::WindowUpdate);
         self.cast()
@@ -101,6 +105,7 @@ impl Header<()> {
         debug_assert_eq!(self.tag, Tag::Ping);
         self.cast()
     }
+
 }
 
 impl<T: HasSyn> Header<T> {
@@ -143,6 +148,14 @@ impl Header<Data> {
             _marker: std::marker::PhantomData,
         }
     }
+
+    /// Returns the length of the body.
+    ///
+    /// The `length` field in the header has a different semantic meaning depending on the tag.
+    /// For [`Tag::Data`], it describes the length of the body.
+    pub fn body_len(&self) -> usize {
+        self.length.val() as usize
+    }
 }
 
 impl Header<WindowUpdate> {