tests pass

Cargo.toml

@@ -16,7 +16,6 @@ ordermap = "0.2.10"
 
 [dev-dependencies]
 mock-io = { git = "https://github.com/carllerche/mock-io" }
-test_futures = "0.0.1"
 
 # Fuzzing
 quickcheck = "0.4.1"

src/proto/connection.rs

@@ -72,6 +72,7 @@ impl<T, P, B> Connection<T, P, B>
                      end_of_stream: bool)
         -> sink::Send<Self>
     {
+        trace!("send_data: id={:?}", id);
         self.send(Frame::Data {
             id,
             data,

src/proto/flow_control.rs

@@ -407,12 +407,12 @@ impl<T: ControlStreams> ControlStreams for FlowControl<T> {
         self.inner.send_flow_controller(id)
     }
 
-    fn check_can_send_data(&mut self, id: StreamId) -> Result<(), ConnectionError> {
+    fn can_send_data(&mut self, id: StreamId) -> bool {
-        self.inner.check_can_send_data(id)
+        self.inner.can_send_data(id)
     }
 
-    fn check_can_recv_data(&mut self, id: StreamId) -> Result<(), ConnectionError>  {
+    fn can_recv_data(&mut self, id: StreamId) -> bool  {
-        self.inner.check_can_recv_data(id)
+        self.inner.can_recv_data(id)
     }
 }
 

src/proto/framed_write.rs

@@ -155,7 +155,7 @@ impl<T, B> Sink for FramedWrite<T, B>
     }
 
     fn poll_complete(&mut self) -> Poll<(), ConnectionError> {
-        trace!("FramedWrite::poll_complete");
+        trace!("poll_complete");
 
         // TODO: implement
         match self.next {
@@ -165,6 +165,7 @@ impl<T, B> Sink for FramedWrite<T, B>
 
         // As long as there is data to write, try to write it!
         while !self.is_empty() {
+            trace!("writing {}", self.buf.remaining());
             try_ready!(self.inner.write_buf(&mut self.buf));
         }
 

src/proto/state.rs

@@ -1,6 +1,5 @@
 use ConnectionError;
 use error::Reason::*;
-use error::User::*;
 use proto::{FlowControlState, WindowSize};
 
 /// Represents the state of an H2 stream
@@ -156,40 +155,24 @@ impl StreamState {
         Ok(true)
     }
 
-    pub fn check_can_send_data(&self) -> Result<(), ConnectionError> {
+    pub fn can_send_data(&self) -> bool {
         use self::StreamState::*;
         match self {
-            &Open { ref remote, .. } => {
+            &Idle | &Closed | &HalfClosedRemote(..) => false,
-                try!(remote.check_streaming(UnexpectedFrameType.into()));
-                Ok(())
-            }
 
-            &HalfClosedLocal(ref remote) => {
+            &Open { ref remote, .. } |
-                try!(remote.check_streaming(UnexpectedFrameType.into()));
+            &HalfClosedLocal(ref remote) => remote.is_streaming(),
-                Ok(())
-            }
-
-            &Idle | &Closed | &HalfClosedRemote(..) => {
-                Err(UnexpectedFrameType.into())
-            }
         }
     }
 
-    pub fn check_can_recv_data(&self) -> Result<(), ConnectionError> {
+    pub fn can_recv_data(&self) -> bool {
         use self::StreamState::*;
         match self {
-            &Open { ref local, .. } => {
+            &Idle | &Closed | &HalfClosedLocal(..) => false,
-                try!(local.check_streaming(ProtocolError.into()));
-                Ok(())
-            }
 
+            &Open { ref local, .. } |
             &HalfClosedRemote(ref local) => {
-                try!(local.check_streaming(ProtocolError.into()));
+                local.is_streaming()
-                Ok(())
-            }
-
-            &Idle | &Closed | &HalfClosedLocal(..) => {
-                Err(ProtocolError.into())
             }
         }
     }
@@ -288,11 +271,11 @@ impl PeerState {
     }
 
     #[inline]
-    fn check_streaming(&self, err: ConnectionError) -> Result<(), ConnectionError> {
+    fn is_streaming(&self) -> bool {
         use self::PeerState::*;
         match self {
-            &Streaming(..) => Ok(()),
+            &Streaming(..) => true,
-            _ => Err(err),
+            _ => false,
         }
     }
 

src/proto/stream_recv_close.rs

@@ -157,12 +157,12 @@ impl<T: ControlStreams> ControlStreams for StreamRecvClose<T> {
         self.inner.send_flow_controller(id)
     }
 
-    fn check_can_send_data(&mut self, id: StreamId) -> Result<(), ConnectionError> {
+    fn can_send_data(&mut self, id: StreamId) -> bool {
-        self.inner.check_can_send_data(id)
+        self.inner.can_send_data(id)
     }
 
-    fn check_can_recv_data(&mut self, id: StreamId) -> Result<(), ConnectionError>  {
+    fn can_recv_data(&mut self, id: StreamId) -> bool  {
-        self.inner.check_can_recv_data(id)
+        self.inner.can_recv_data(id)
     }
 }
 

src/proto/stream_recv_open.rs

@@ -170,7 +170,9 @@ impl<T, U> Stream for StreamRecvOpen<T>
 
             if let &Data(..) = &frame {
                 // Ensures we've already received headers for this stream.
-                self.inner.check_can_recv_data(id)?;
+                if !self.inner.can_recv_data(id) {
+                    return Err(ProtocolError.into());
+                }
             }
 
             // If the frame ends the stream, it will be handled in
@@ -400,12 +402,12 @@ impl<T: ControlStreams> ControlStreams for StreamRecvOpen<T> {
         self.inner.send_flow_controller(id)
     }
 
-    fn check_can_send_data(&mut self, id: StreamId) -> Result<(), ConnectionError> {
+    fn can_send_data(&mut self, id: StreamId) -> bool {
-        self.inner.check_can_send_data(id)
+        self.inner.can_send_data(id)
     }
 
-    fn check_can_recv_data(&mut self, id: StreamId) -> Result<(), ConnectionError>  {
+    fn can_recv_data(&mut self, id: StreamId) -> bool  {
-        self.inner.check_can_recv_data(id)
+        self.inner.can_recv_data(id)
     }
 }
 

src/proto/stream_send_close.rs

@@ -162,12 +162,12 @@ impl<T: ControlStreams> ControlStreams for StreamSendClose<T> {
         self.inner.send_flow_controller(id)
     }
 
-    fn check_can_send_data(&mut self, id: StreamId) -> Result<(), ConnectionError> {
+    fn can_send_data(&mut self, id: StreamId) -> bool {
-        self.inner.check_can_send_data(id)
+        self.inner.can_send_data(id)
     }
 
-    fn check_can_recv_data(&mut self, id: StreamId) -> Result<(), ConnectionError>  {
+    fn can_recv_data(&mut self, id: StreamId) -> bool  {
-        self.inner.check_can_recv_data(id)
+        self.inner.can_recv_data(id)
     }
 }
 

src/proto/stream_send_open.rs

@@ -1,5 +1,5 @@
 use ConnectionError;
-use error::User::{InvalidStreamId, StreamReset, Rejected};
+use error::User::{InactiveStreamId, InvalidStreamId, StreamReset, Rejected, UnexpectedFrameType};
 use frame::{Frame, SettingSet};
 use proto::*;
 
@@ -103,7 +103,11 @@ impl<T, U> Sink for StreamSendOpen<T>
         }
 
         if T::local_valid_id(id) {
-            if !self.inner.is_local_active(id) {
+            if self.inner.is_local_active(id) {
+                if !self.inner.can_send_data(id) {
+                    return Err(InactiveStreamId.into());
+                }
+            } else {
                 if !T::local_can_open() {
                     return Err(InvalidStreamId.into());
                 }
@@ -114,8 +118,11 @@ impl<T, U> Sink for StreamSendOpen<T>
                     }
                 }
 
-                trace!("creating new local stream");
+                if let &Frame::Headers(..) = &frame {
-                self.inner.local_open(id, self.initial_window_size)?;
+                    self.inner.local_open(id, self.initial_window_size)?;
+                } else {
+                    return Err(InactiveStreamId.into());
+                }
             }
         } else {
             // If the frame is part of a remote stream, it MUST already exist.
@@ -130,7 +137,9 @@ impl<T, U> Sink for StreamSendOpen<T>
 
         if let &Frame::Data(..) = &frame {
             // Ensures we've already sent headers for this stream.
-            self.inner.check_can_send_data(id)?;
+            if !self.inner.can_send_data(id) {
+                return Err(InactiveStreamId.into());
+            }
         }
 
         trace!("sending frame...");
@@ -230,12 +239,12 @@ impl<T: ControlStreams> ControlStreams for StreamSendOpen<T> {
         self.inner.send_flow_controller(id)
     }
 
-    fn check_can_send_data(&mut self, id: StreamId) -> Result<(), ConnectionError> {
+    fn can_send_data(&mut self, id: StreamId) -> bool {
-        self.inner.check_can_send_data(id)
+        self.inner.can_send_data(id)
     }
 
-    fn check_can_recv_data(&mut self, id: StreamId) -> Result<(), ConnectionError>  {
+    fn can_recv_data(&mut self, id: StreamId) -> bool  {
-        self.inner.check_can_recv_data(id)
+        self.inner.can_recv_data(id)
     }
 }
 

src/proto/stream_store.rs

@@ -76,8 +76,8 @@ pub trait ControlStreams {
     fn update_inital_recv_window_size(&mut self, old_sz: u32, new_sz: u32);
     fn update_inital_send_window_size(&mut self, old_sz: u32, new_sz: u32);
 
-    fn check_can_send_data(&mut self, id: StreamId) -> Result<(), ConnectionError>;
+    fn can_send_data(&mut self, id: StreamId) -> bool;
-    fn check_can_recv_data(&mut self, id: StreamId) -> Result<(), ConnectionError>;
+    fn can_recv_data(&mut self, id: StreamId) -> bool;
 }
 
 /// Holds the underlying stream state to be accessed by upper layers.
@@ -366,18 +366,18 @@ impl<T, P: Peer> ControlStreams for StreamStore<T, P> {
         }
     }
 
-    fn check_can_send_data(&mut self, id: StreamId) -> Result<(), ConnectionError> {
+    fn can_send_data(&mut self, id: StreamId) -> bool {
-        if let Some(s) = self.get_active(id) {
+        match self.get_active(id) {
-            return s.check_can_send_data();
+            Some(s) => s.can_send_data(),
+            None => false,
         }
-        Err(ProtocolError.into())
     }
 
-    fn check_can_recv_data(&mut self, id: StreamId) -> Result<(), ConnectionError>  {
+    fn can_recv_data(&mut self, id: StreamId) -> bool  {
-        if let Some(s) = self.get_active(id) {
+        match self.get_active(id) {
-            return s.check_can_recv_data();
+            Some(s) => s.can_recv_data(),
+            None => false,
         }
-        Err(ProtocolError.into())
     }
 }
 

tests/client_request.rs

@@ -6,7 +6,6 @@ extern crate futures;
 #[macro_use]
 extern crate log;
 extern crate mock_io;
-extern crate test_futures;
 
 // scoped so `cargo test client_request` dtrt.
 mod client_request {
@@ -16,7 +15,6 @@ mod client_request {
     use futures::*;
     use bytes::Bytes;
     use mock_io;
-    use test_futures::*;
 
     // TODO: move into another file
     macro_rules! assert_user_err {
@@ -50,12 +48,12 @@ mod client_request {
             .write(SETTINGS_ACK)
             .build();
 
-        let mut h2 = client::handshake(mock)
+        let h2 = client::handshake(mock)
             .wait().unwrap();
         trace!("hands have been shook");
 
         // At this point, the connection should be closed
-        sassert_done(&mut h2);
+        assert!(Stream::wait(h2).next().is_none());
     }
 
     #[test]
@@ -87,7 +85,7 @@ mod client_request {
         // Get the response
 
         trace!("getting response");
-        let (resp, mut h2) = h2.into_future().wait().unwrap();
+        let (resp, h2) = h2.into_future().wait().unwrap();
 
         match resp.unwrap() {
             Frame::Headers { headers, .. } => {
@@ -98,7 +96,7 @@ mod client_request {
 
         // No more frames
         trace!("ensure no more responses");
-        sassert_done(&mut h2);
+        assert!(Stream::wait(h2).next().is_none());;
     }
 
     #[test]
@@ -151,7 +149,7 @@ mod client_request {
         }
 
         // Get the response body
-        let (data, mut h2) = h2.into_future().wait().unwrap();
+        let (data, h2) = h2.into_future().wait().unwrap();
 
         match data.unwrap() {
             Frame::Data { id, data, end_of_stream, .. } => {
@@ -162,7 +160,7 @@ mod client_request {
             _ => panic!("unexpected frame"),
         }
 
-        sassert_done(&mut h2);
+        assert!(Stream::wait(h2).next().is_none());;
     }
 
     #[test]
@@ -215,7 +213,7 @@ mod client_request {
         }
 
         // Get the response body
-        let (data, mut h2) = h2.into_future().wait().unwrap();
+        let (data, h2) = h2.into_future().wait().unwrap();
 
         match data.unwrap() {
             Frame::Data { id, data, end_of_stream, .. } => {
@@ -226,7 +224,7 @@ mod client_request {
             _ => panic!("unexpected frame"),
         }
 
-        sassert_done(&mut h2);
+        assert!(Stream::wait(h2).next().is_none());;
     }
 
     #[test]
@@ -295,7 +293,7 @@ mod client_request {
         }
 
         // Get the response body
-        let (data, mut h2) = h2.into_future().wait().expect("into future");
+        let (data, h2) = h2.into_future().wait().expect("into future");
 
         match data.expect("response data") {
             Frame::Data { id, data, end_of_stream, .. } => {
@@ -306,7 +304,7 @@ mod client_request {
             _ => panic!("unexpected frame"),
         }
 
-        sassert_done(&mut h2);
+        assert!(Stream::wait(h2).next().is_none());;
     }
 
     #[test]
@@ -352,7 +350,7 @@ mod client_request {
         request.uri = "https://http2.akamai.com/".parse().unwrap();
         let err = h2.send_request(1.into(), request, true).wait().unwrap_err();
 
-        assert_user_err!(err, UnexpectedFrameType);
+        assert_user_err!(err, InactiveStreamId);
     }
 
     #[test]
@@ -399,7 +397,7 @@ mod client_request {
 
         // Send the data
         let err = h2.send_data(id, body.into(), true).wait().unwrap_err();
-        assert_user_err!(err, UnexpectedFrameType);
+        assert_user_err!(err, InactiveStreamId);
     }
 
     #[test]