Wire in recv flow control

src/client.rs

@@ -1,5 +1,5 @@
 use {frame, ConnectionError, StreamId};
-use {Body, Chunk};
+use Body;
 use proto::{self, Connection, WindowSize};
 use error::Reason::*;
 

src/frame/go_away.rs

@@ -3,13 +3,20 @@ use frame::{self, Head, Error, Kind, StreamId};
 
 use bytes::{BufMut, BigEndian};
 
-#[derive(Debug)]
+#[derive(Debug, Clone, Copy)]
 pub struct GoAway {
     last_stream_id: StreamId,
     error_code: u32,
 }
 
 impl GoAway {
+    pub fn new(last_stream_id: StreamId, reason: Reason) -> Self {
+        GoAway {
+            last_stream_id,
+            error_code: reason.into(),
+        }
+    }
+
     pub fn reason(&self) -> Reason {
         self.error_code.into()
     }

src/frame/mod.rs

@@ -165,6 +165,9 @@ pub enum Error {
     /// An invalid setting value was provided
     InvalidSettingValue,
 
+    /// An invalid window update value
+    InvalidWindowUpdateValue,
+
     /// The payload length specified by the frame header was not the
     /// value necessary for the specific frame type.
     InvalidPayloadLength,

src/frame/window_update.rs

@@ -38,7 +38,9 @@ impl WindowUpdate {
         // when received.
         let size_increment = unpack_octets_4!(payload, 0, u32) & !SIZE_INCREMENT_MASK;
 
-        // TODO: the size_increment must be greater than 0
+        if size_increment == 0 {
+            return Err(Error::InvalidWindowUpdateValue.into());
+        }
 
         Ok(WindowUpdate {
             stream_id: head.stream_id(),

src/lib.rs

@@ -53,30 +53,14 @@ pub struct Body<B: IntoBuf> {
     inner: proto::StreamRef<B::Buf>,
 }
 
-#[derive(Debug)]
-pub struct Chunk<B: IntoBuf> {
-    inner: proto::Chunk<B::Buf>,
-}
-
 // ===== impl Body =====
 
 impl<B: IntoBuf> futures::Stream for Body<B> {
-    type Item = Chunk<B>;
+    type Item = Bytes;
     type Error = ConnectionError;
 
     fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
-        let chunk = try_ready!(self.inner.poll_data())
-            .map(|inner| Chunk { inner });
-
-        Ok(chunk.into())
-    }
-}
-
-// ===== impl Chunk =====
-
-impl<B: IntoBuf> Chunk<B> {
-    pub fn pop_bytes(&mut self) -> Option<Bytes> {
-        self.inner.pop_bytes()
+        self.inner.poll_data()
     }
 }
 

src/proto/connection.rs

@@ -13,14 +13,40 @@ use std::marker::PhantomData;
 /// An H2 connection
 #[derive(Debug)]
 pub(crate) struct Connection<T, P, B: IntoBuf = Bytes> {
-    // Codec
+    /// Tracks the connection level state transitions.
+    state: State,
+
+    /// Read / write frame values
     codec: Codec<T, Prioritized<B::Buf>>,
+
+    /// Ping/pong handler
     ping_pong: PingPong<Prioritized<B::Buf>>,
+
+    /// Connection settings
     settings: Settings,
+
+    /// Stream state handler
     streams: Streams<B::Buf>,
+
+    /// Client or server
     _phantom: PhantomData<P>,
 }
 
+#[derive(Debug)]
+enum State {
+    /// Currently open in a sane state
+    Open,
+
+    /// Waiting to send a GO_AWAY frame
+    GoAway(frame::GoAway),
+
+    /// The codec must be flushed
+    Flush(Reason),
+
+    /// In an errored state
+    Error(Reason),
+}
+
 impl<T, P, B> Connection<T, P, B>
     where T: AsyncRead + AsyncWrite,
           P: Peer,
@@ -36,6 +62,7 @@ impl<T, P, B> Connection<T, P, B>
         });
 
         Connection {
+            state: State::Open,
             codec: codec,
             ping_pong: PingPong::new(),
             settings: Settings::new(),
@@ -62,13 +89,36 @@ impl<T, P, B> Connection<T, P, B>
 
     /// Advances the internal state of the connection.
     pub fn poll(&mut self) -> Poll<(), ConnectionError> {
-        match self.poll2() {
-            Err(e) => {
-                debug!("Connection::poll; err={:?}", e);
-                self.streams.recv_err(&e);
-                Err(e)
+        use error::ConnectionError::*;
+
+        loop {
+            match self.state {
+                // When open, continue to poll a frame
+                State::Open => {},
+                // In an error state
+                _ => {
+                    try_ready!(self.poll_complete());
+
+                    // GO_AWAY frame has been sent, return the error
+                    return Err(self.state.error().unwrap().into());
+                }
+            }
+
+            match self.poll2() {
+                Err(Proto(e)) => {
+                    debug!("Connection::poll; err={:?}", e);
+                    let last_processed_id = self.streams.recv_err(&e.into());
+                    let frame = frame::GoAway::new(last_processed_id, e);
+
+                    self.state = State::GoAway(frame);
+                }
+                Err(e) => {
+                    // TODO: Are I/O errors recoverable?
+                    self.streams.recv_err(&e);
+                    return Err(e);
+                }
+                ret => return ret,
             }
-            ret => ret,
         }
     }
 
@@ -114,7 +164,7 @@ impl<T, P, B> Connection<T, P, B>
                     self.settings.recv_settings(frame);
                 }
                 Some(GoAway(frame)) => {
-                    // TODO: handle the last_stream_id. Also, should this be
+                    // TODO: handle the last_processed_id. Also, should this be
                     // handled as an error?
                     let e = ConnectionError::Proto(frame.reason());
                     return Ok(().into());
@@ -141,12 +191,34 @@ impl<T, P, B> Connection<T, P, B>
     }
 
     fn poll_complete(&mut self) -> Poll<(), ConnectionError> {
-        try_ready!(self.poll_ready());
+        loop {
+            match self.state {
+                State::Open => {
+                    try_ready!(self.poll_ready());
 
-        // Ensure all window updates have been sent.
-        try_ready!(self.streams.poll_complete(&mut self.codec));
+                    // Ensure all window updates have been sent.
+                    try_ready!(self.streams.poll_complete(&mut self.codec));
 
-        Ok(().into())
+                    return Ok(().into());
+                }
+                State::GoAway(frame) => {
+                    if !self.codec.start_send(frame.into())?.is_ready() {
+                        // Not ready to send the frame... try again later.
+                        return Ok(Async::NotReady);
+                    }
+
+                    // GO_AWAY sent, transition the connection to an errored state
+                    self.state = State::Flush(frame.reason());
+                }
+                State::Flush(reason) => {
+                    try_ready!(self.codec.poll_complete());
+                    self.state = State::Error(reason);
+                }
+                State::Error(..) => {
+                    return Ok(().into());
+                }
+            }
+        }
     }
 
     fn convert_poll_message(frame: frame::Headers) -> Result<Frame<P::Poll>, ConnectionError> {
@@ -185,3 +257,21 @@ impl<T, B> Connection<T, server::Peer, B>
         self.streams.next_incoming()
     }
 }
+
+// ====== impl State =====
+
+impl State {
+    fn is_open(&self) -> bool {
+        match *self {
+            State::Open => true,
+            _ => false,
+        }
+    }
+
+    fn error(&self) -> Option<Reason> {
+        match *self {
+            State::Error(reason) => Some(reason),
+            _ => None,
+        }
+    }
+}

src/proto/mod.rs

@@ -7,7 +7,7 @@ mod settings;
 mod streams;
 
 pub(crate) use self::connection::Connection;
-pub(crate) use self::streams::{Streams, StreamRef, Chunk};
+pub(crate) use self::streams::{Streams, StreamRef};
 
 use self::codec::Codec;
 use self::framed_read::FramedRead;
@@ -21,6 +21,7 @@ use error::Reason;
 use frame::{self, Frame};
 
 use futures::{self, task, Poll, Async, AsyncSink, Sink, Stream as Stream2};
+use futures::task::Task;
 use bytes::{Buf, IntoBuf};
 use tokio_io::{AsyncRead, AsyncWrite};
 use tokio_io::codec::length_delimited;
@@ -57,7 +58,7 @@ pub struct WindowUpdate {
 
 // Constants
 pub const DEFAULT_INITIAL_WINDOW_SIZE: WindowSize = 65_535;
-pub const MAX_WINDOW_SIZE: WindowSize = ::std::u32::MAX;
+pub const MAX_WINDOW_SIZE: WindowSize = (1 << 31) - 1;
 
 /// Create a transport prepared to handle the server handshake.
 ///

src/proto/streams/buffer.rs

@@ -108,53 +108,4 @@ impl<B> Deque<B> {
             None => None,
         }
     }
-
-    pub fn take_while<F>(&mut self, buf: &mut Buffer<B>, mut f: F) -> Self
-        where F: FnMut(&Frame<B>) -> bool
-    {
-        match self.indices {
-            Some(mut idxs) => {
-                if !f(&buf.slab[idxs.head].frame) {
-                    return Deque::new();
-                }
-
-                let head = idxs.head;
-                let mut tail = idxs.head;
-
-                loop {
-                    let next = match buf.slab[tail].next {
-                        Some(next) => next,
-                        None => {
-                            self.indices = None;
-                            return Deque {
-                                indices: Some(idxs),
-                                _p: PhantomData,
-                            };
-                        }
-                    };
-
-                    if !f(&buf.slab[next].frame) {
-                        // Split the linked list
-                        buf.slab[tail].next = None;
-
-                        self.indices = Some(Indices {
-                            head: next,
-                            tail: idxs.tail,
-                        });
-
-                        return Deque {
-                            indices: Some(Indices {
-                                head: head,
-                                tail: tail,
-                            }),
-                            _p: PhantomData,
-                        }
-                    }
-
-                    tail = next;
-                }
-            }
-            None => Deque::new(),
-        }
-    }
 }