ensure that the length delimited codec expects the right packet size

this was causing errors because we were buffering up to 10MB of terminal
data, but by default, the length delimited codec throws an error if it
sees a packet of over 8MB. this drops the default buffer size to 4MB,
and also ensures that if the buffer size changes that we update the
codec max packet size to match (with some extra to handle overhead and
other packet types).

1 files changed, 7 insertions, 5 deletions

diff --git a/src/protocol.rs b/src/protocol.rs
index e781e94..70c9ac3 100644
--- a/src/protocol.rs
+++ b/src/protocol.rs
@@ -71,10 +71,11 @@ pub struct FramedReader<T: tokio::io::AsyncRead>(
 );
 
 impl<T: tokio::io::AsyncRead> FramedReader<T> {
-    pub fn new(rs: T) -> Self {
+    pub fn new(rs: T, buffer_size: usize) -> Self {
         Self(
             tokio::codec::length_delimited::Builder::new()
                 .length_field_length(4)
+                .max_frame_length(buffer_size + 1024 * 1024)
                 .new_read(rs),
         )
     }
@@ -88,10 +89,11 @@ pub struct FramedWriter<T: tokio::io::AsyncWrite>(
 );
 
 impl<T: tokio::io::AsyncWrite> FramedWriter<T> {
-    pub fn new(ws: T) -> Self {
+    pub fn new(ws: T, buffer_size: usize) -> Self {
         Self(
             tokio::codec::length_delimited::Builder::new()
                 .length_field_length(4)
+                .max_frame_length(buffer_size + 1024 * 1024)
                 .new_write(ws),
         )
     }
@@ -600,11 +602,11 @@ mod test {
             let wres2 = wres.clone();
             let buf = std::io::Cursor::new(vec![]);
             let fut = msg
-                .write_async(FramedWriter::new(buf))
+                .write_async(FramedWriter::new(buf, 4_194_304))
                 .and_then(|w| {
                     let mut buf = w.0.into_inner();
                     buf.set_position(0);
-                    Message::read_async(FramedReader::new(buf))
+                    Message::read_async(FramedReader::new(buf, 4_194_304))
                 })
                 .and_then(move |(msg2, _)| {
                     wres.wait().send(Ok(msg2)).unwrap();
@@ -636,7 +638,7 @@ mod test {
             let (wres, rres) = tokio::sync::mpsc::channel(1);
             let wres2 = wres.clone();
             let buf = std::io::Cursor::new(buf);
-            let fut = Message::read_async(FramedReader::new(buf))
+            let fut = Message::read_async(FramedReader::new(buf, 4_194_304))
                 .and_then(move |(msg2, _)| {
                     wres.wait().send(Ok(msg2)).unwrap();
                     futures::future::ok(())