Merge branch 'master' into ver/flowio

src/proto/connection.rs

@@ -110,6 +110,20 @@ impl<T, P> Stream for Connection<T, P>
                 let stream_id = v.stream_id();
                 let end_of_stream = v.is_end_stream();
 
+                let stream_initialized = try!(self.streams.entry(stream_id)
+                     .or_insert(State::default())
+                     .recv_headers::<P>(end_of_stream));
+
+                if stream_initialized {
+                    // TODO: Ensure available capacity for a new stream
+                    // This won't be as simple as self.streams.len() as closed
+                    // connections should not be factored.
+
+                    if !P::is_valid_remote_stream_id(stream_id) {
+                        unimplemented!();
+                    }
+                }
+
                 Frame::Headers {
                     id: stream_id,
                     headers: P::convert_poll_message(v),
@@ -153,22 +167,23 @@ impl<T, P> Sink for Connection<T, P>
 
         match item {
             Frame::Headers { id, headers, end_of_stream } => {
-                // Ensure ID is valid
-                // TODO: This check should only be done **if** this is a new
-                // stream ID
-                // try!(P::check_initiating_id(id));
-
-                // TODO: Ensure available capacity for a new stream
-                // This won't be as simple as self.streams.len() as closed
-                // connections should not be factored.
-
                 // Transition the stream state, creating a new entry if needed
                 //
                 // TODO: Response can send multiple headers frames before body
                 // (1xx responses).
-                try!(self.streams.entry(id)
+                let stream_initialized = try!(self.streams.entry(id)
                      .or_insert(State::default())
-                     .send_headers());
+                     .send_headers::<P>(end_of_stream));
+
+                if stream_initialized {
+                    // TODO: Ensure available capacity for a new stream
+                    // This won't be as simple as self.streams.len() as closed
+                    // connections should not be factored.
+                    //
+                    if !P::is_valid_local_stream_id(id) {
+                        unimplemented!();
+                    }
+                }
 
                 let frame = P::convert_send_message(id, headers, end_of_stream);
 

src/proto/state.rs

@@ -1,4 +1,4 @@
-use ConnectionError;
+use {ConnectionError, Reason, Peer};
 
 /// Represents the state of an H2 stream
 ///
@@ -45,23 +45,134 @@ pub enum State {
     Idle,
     ReservedLocal,
     ReservedRemote,
-    Open,
-    HalfClosedLocal,
-    HalfClosedRemote,
+    Open {
+        local: PeerState,
+        remote: PeerState,
+    },
+    HalfClosedLocal(PeerState),
+    HalfClosedRemote(PeerState),
     Closed,
 }
 
+#[derive(Debug, Copy, Clone, Eq, PartialEq)]
+pub enum PeerState {
+    Headers,
+    Data,
+}
+
 impl State {
+    /// Transition the state to represent headers being received.
+    ///
+    /// Returns true if this state transition results in iniitializing the
+    /// stream id. `Err` is returned if this is an invalid state transition.
+    pub fn recv_headers<P: Peer>(&mut self, eos: bool) -> Result<bool, ConnectionError> {
+        use self::State::*;
+        use self::PeerState::*;
+
+        match *self {
+            Idle => {
+                *self = if eos {
+                    HalfClosedRemote(Headers)
+                } else {
+                    Open {
+                        local: Headers,
+                        remote: Data,
+                    }
+                };
+
+                Ok(true)
+            }
+            Open { local, remote } => {
+                try!(remote.check_is_headers(Reason::ProtocolError));
+
+                *self = if eos {
+                    HalfClosedRemote(local)
+                } else {
+                    let remote = Data;
+                    Open { local, remote }
+                };
+
+                Ok(false)
+            }
+            HalfClosedLocal(remote) => {
+                try!(remote.check_is_headers(Reason::ProtocolError));
+
+                *self = if eos {
+                    Closed
+                } else {
+                    HalfClosedLocal(Data)
+                };
+
+                Ok(false)
+            }
+            Closed | HalfClosedRemote(..) => {
+                Err(Reason::ProtocolError.into())
+            }
+            _ => unimplemented!(),
+        }
+    }
+
     /// Transition the state to represent headers being sent.
     ///
-    /// Returns an error if this is an invalid state transition.
-    pub fn send_headers(&mut self) -> Result<(), ConnectionError> {
-        if *self != State::Idle {
-            unimplemented!();
-        }
+    /// Returns true if this state transition results in initializing the stream
+    /// id. `Err` is returned if this is an invalid state transition.
+    pub fn send_headers<P: Peer>(&mut self, eos: bool) -> Result<bool, ConnectionError> {
+        use self::State::*;
+        use self::PeerState::*;
 
-        *self = State::Open;
-        Ok(())
+        match *self {
+            Idle => {
+                *self = if eos {
+                    HalfClosedLocal(Headers)
+                } else {
+                    Open {
+                        local: Data,
+                        remote: Headers,
+                    }
+                };
+
+                Ok(true)
+            }
+            Open { local, remote } => {
+                try!(local.check_is_headers(Reason::InternalError));
+
+                *self = if eos {
+                    HalfClosedLocal(remote)
+                } else {
+                    let local = Data;
+                    Open { local, remote }
+                };
+
+                Ok(false)
+            }
+            HalfClosedRemote(local) => {
+                try!(local.check_is_headers(Reason::InternalError));
+
+                *self = if eos {
+                    Closed
+                } else {
+                    HalfClosedRemote(Data)
+                };
+
+                Ok(false)
+            }
+            Closed | HalfClosedLocal(..) => {
+                Err(Reason::InternalError.into())
+            }
+            _ => unimplemented!(),
+        }
+    }
+}
+
+impl PeerState {
+    #[inline]
+    fn check_is_headers(&self, err: Reason) -> Result<(), ConnectionError> {
+        use self::PeerState::*;
+
+        match *self {
+            Headers => Ok(()),
+            _ => Err(err.into()),
+        }
     }
 }