RUST-269 Wrap I/O in buffered stream (#113)

Author: saghm

Cargo.toml

@@ -14,6 +14,7 @@ version = "0.9.0"
 base64 = "0.11.0"
 bitflags = "1.1.0"
 bson = "0.14.0"
+bufstream = "0.1.4"
 byteorder = "1.3.2"
 chrono = "0.4.7"
 derivative = "1.0.2"

src/cmap/conn/stream.rs

@@ -5,6 +5,7 @@ use std::{
     time::Duration,
 };
 
+use bufstream::BufStream;
 use derivative::Derivative;
 use webpki::DNSNameRef;
 
@@ -28,10 +29,13 @@ pub(super) enum Stream {
     Null,
 
     /// A basic TCP connection to the server.
-    Tcp(TcpStream),
+    Tcp(BufStream<TcpStream>),
 
     /// A TLS connection over TCP.
-    Tls(#[derivative(Debug = "ignore")] rustls::StreamOwned<rustls::ClientSession, TcpStream>),
+    Tls(
+        #[derivative(Debug = "ignore")]
+        BufStream<rustls::StreamOwned<rustls::ClientSession, TcpStream>>,
+    ),
 }
 
 impl Stream {
@@ -54,6 +58,7 @@ impl Stream {
             TcpStream::connect_timeout(&socket_addrs[0], timeout)?
         };
         inner.set_nodelay(true)?;
+        let inner = inner;
 
         match tls_options {
             Some(cfg) => {
@@ -63,9 +68,11 @@ impl Stream {
 
                 let session = rustls::ClientSession::new(&Arc::new(tls_config), name);
 
-                Ok(Stream::Tls(rustls::StreamOwned::new(session, inner)))
+                Ok(Stream::Tls(BufStream::new(rustls::StreamOwned::new(
+                    session, inner,
+                ))))
             }
-            None => Ok(Self::Tcp(inner)),
+            None => Ok(Self::Tcp(BufStream::new(inner))),
         }
     }
 }