Bunch of work

src/proto/framed_read.rs

@@ -1,5 +1,6 @@
-use ConnectionError;
+use {hpack, ConnectionError};
 use frame::{self, Frame, Kind};
+use frame::DEFAULT_SETTINGS_HEADER_TABLE_SIZE;
 
 use tokio_io::AsyncWrite;
 
@@ -12,7 +13,7 @@ pub struct FramedRead<T> {
     inner: T,
 
     // hpack decoder state
-    // hpack: hpack::Decoder,
+    hpack: hpack::Decoder,
 
 }
 
@@ -23,6 +24,7 @@ impl<T> FramedRead<T>
     pub fn new(inner: T) -> FramedRead<T> {
         FramedRead {
             inner: inner,
+            hpack: hpack::Decoder::new(DEFAULT_SETTINGS_HEADER_TABLE_SIZE),
         }
     }
 }
@@ -46,10 +48,7 @@ impl<T> FramedRead<T> {
             Kind::GoAway => unimplemented!(),
             Kind::WindowUpdate => unimplemented!(),
             Kind::Continuation => unimplemented!(),
-            Kind::Unknown => {
-                let _ = bytes.split_to(frame::HEADER_LEN);
-                frame::Unknown::new(head, bytes).into()
-            }
+            Kind::Unknown => return Ok(None),
         };
 
         Ok(Some(frame))

src/proto/framed_write.rs

@@ -1,33 +1,86 @@
 use {ConnectionError, Reason};
-use frame::{Frame, Error};
+use frame::{self, Data, Frame, Error, Headers, PushPromise, Settings};
+use hpack;
 
-use tokio_io::AsyncWrite;
 use futures::*;
-use bytes::{BytesMut, Buf, BufMut};
+use tokio_io::AsyncWrite;
+use bytes::{Bytes, BytesMut, Buf, BufMut};
+use http::header::{self, HeaderValue};
 
+use std::cmp;
 use std::io::{self, Cursor};
 
 #[derive(Debug)]
 pub struct FramedWrite<T> {
+    /// Upstream `AsyncWrite`
     inner: T,
-    buf: Cursor<BytesMut>,
+
+    /// HPACK encoder
+    hpack: hpack::Encoder,
+
+    /// Write buffer
+    buf: BytesMut,
+
+    /// Position in the frame
+    pos: usize,
+
+    /// Next frame to encode
+    next: Option<Next>,
+
+    /// Max frame size, this is specified by the peer
+    max_frame_size: usize,
 }
 
-const DEFAULT_BUFFER_CAPACITY: usize = 8 * 1_024;
-const MAX_BUFFER_CAPACITY: usize = 16 * 1_024;
+#[derive(Debug)]
+enum Next {
+    Data {
+        /// Length of the current frame being written
+        frame_len: usize,
+
+        /// Data frame to encode
+        data: frame::Data
+    },
+    Continuation {
+        /// Stream ID of continuation frame
+        stream_id: frame::StreamId,
+
+        /// Argument to pass to the HPACK encoder to resume encoding
+        resume: hpack::EncodeState,
+
+        /// remaining headers to encode
+        rem: header::IntoIter<HeaderValue>,
+    },
+}
+
+/// Initialze the connection with this amount of write buffer.
+const DEFAULT_BUFFER_CAPACITY: usize = 4 * 1_024;
+
+/// Min buffer required to attempt to write a frame
+const MIN_BUFFER_CAPACITY: usize = frame::HEADER_LEN + CHAIN_THRESHOLD;
+
+/// Chain payloads bigger than this. The remote will never advertise a max frame
+/// size less than this (well, the spec says the max frame size can't be less
+/// than 16kb, so not even close).
+const CHAIN_THRESHOLD: usize = 256;
 
 impl<T: AsyncWrite> FramedWrite<T> {
     pub fn new(inner: T) -> FramedWrite<T> {
-        let buf = BytesMut::with_capacity(DEFAULT_BUFFER_CAPACITY);
-
         FramedWrite {
             inner: inner,
-            buf: Cursor::new(buf),
+            hpack: hpack::Encoder::default(),
+            buf: BytesMut::with_capacity(DEFAULT_BUFFER_CAPACITY),
+            pos: 0,
+            next: None,
+            max_frame_size: frame::DEFAULT_MAX_FRAME_SIZE,
         }
     }
 
-    fn write_buf(&mut self) -> &mut BytesMut {
-        self.buf.get_mut()
+    fn has_capacity(&self) -> bool {
+        self.next.is_none() && self.buf.remaining_mut() >= MIN_BUFFER_CAPACITY
+    }
+
+    fn frame_len(&self, data: &Data) -> usize {
+        cmp::min(self.max_frame_size, data.len())
     }
 }
 
@@ -36,50 +89,49 @@ impl<T: AsyncWrite> Sink for FramedWrite<T> {
     type SinkError = ConnectionError;
 
     fn start_send(&mut self, item: Frame) -> StartSend<Frame, ConnectionError> {
-        let len = item.encode_len();
-
-        if len > MAX_BUFFER_CAPACITY {
-            // This case should never happen. Large frames should be chunked at
-            // a higher level, so this is an internal error.
-            return Err(ConnectionError::Proto(Reason::InternalError));
-        }
-
-        if self.write_buf().remaining_mut() <= len {
-            // Try flushing the buffer
+        if self.has_capacity() {
+            // Try flushing
             try!(self.poll_complete());
 
-            let rem = self.write_buf().remaining_mut();
-            let additional = len - rem;
-
-            if self.write_buf().capacity() + additional > MAX_BUFFER_CAPACITY {
+            if self.has_capacity() {
                 return Ok(AsyncSink::NotReady(item));
             }
-
-            // Grow the buffer
-            self.write_buf().reserve(additional);
         }
 
-        // At this point, the buffer contains enough space
-        item.encode(self.write_buf());
+        match item {
+            Frame::Data(v) => {
+                if v.len() >= CHAIN_THRESHOLD {
+                    let head = v.head();
+                    let len = self.frame_len(&v);
+
+                    // Encode the frame head to the buffer
+                    head.encode(len, &mut self.buf);
+
+                    // Save the data frame
+                    self.next = Some(Next::Data {
+                        frame_len: len,
+                        data: v,
+                    });
+                } else {
+                    v.encode(&mut self.buf);
+                }
+            }
+            Frame::Headers(v) => {
+                unimplemented!();
+            }
+            Frame::PushPromise(v) => {
+                unimplemented!();
+            }
+            Frame::Settings(v) => {
+                v.encode(&mut self.buf);
+            }
+        }
 
         Ok(AsyncSink::Ready)
     }
 
     fn poll_complete(&mut self) -> Poll<(), ConnectionError> {
-        while self.buf.has_remaining() {
-            try_ready!(self.inner.write_buf(&mut self.buf));
-
-            if !self.buf.has_remaining() {
-                // Reset the buffer
-                self.write_buf().clear();
-                self.buf.set_position(0);
-            }
-        }
-
-        // Try flushing the underlying IO
-        try_nb!(self.inner.flush());
-
-        return Ok(Async::Ready(()));
+        unimplemented!();
     }
 
     fn close(&mut self) -> Poll<(), ConnectionError> {