Extend stream api to support data.

src/client.rs

@@ -1,4 +1,4 @@
-use {frame, ConnectionError, StreamId};
+use {frame, BodyType, ConnectionError, StreamId, WindowSize};
 use proto::{self, Connection};
 use error::Reason::*;
 
@@ -76,7 +76,7 @@ impl<T, B> Client<T, B>
     /// Returns `Ready` when the connection can initialize a new HTTP 2.0
     /// stream.
     pub fn poll_ready(&mut self) -> Poll<(), ConnectionError> {
-        unimplemented!();
+        self.connection.poll_ready()
     }
 
     /// Send a request on a new HTTP 2.0 stream
@@ -140,10 +140,14 @@ impl<T, B> fmt::Debug for Handshake<T, B>
 
 impl<B: IntoBuf> Stream<B> {
     /// Receive the HTTP/2.0 response, if it is ready.
-    pub fn poll_response(&mut self) -> Poll<Response<()>, ConnectionError> {
+    pub fn poll_response(&mut self) -> Poll<Response<BodyType>, ConnectionError> {
         self.inner.poll_response()
     }
 
+    pub fn poll_data(&mut self, sz: WindowSize) -> Poll<Option<Bytes>, ConnectionError> {
+        self.inner.poll_data(sz)
+    }
+
     /// Send data
     pub fn send_data(&mut self, data: B, end_of_stream: bool)
         -> Result<(), ConnectionError>
@@ -160,7 +164,7 @@ impl<B: IntoBuf> Stream<B> {
 }
 
 impl<B: IntoBuf> Future for Stream<B> {
-    type Item = Response<()>;
+    type Item = Response<BodyType>;
     type Error = ConnectionError;
 
     fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
@@ -172,7 +176,7 @@ impl<B: IntoBuf> Future for Stream<B> {
 
 impl proto::Peer for Peer {
     type Send = Request<()>;
-    type Poll = Response<()>;
+    type Poll = Response<BodyType>;
 
     fn is_server() -> bool {
         false
@@ -202,8 +206,17 @@ impl proto::Peer for Peer {
     }
 
     fn convert_poll_message(headers: frame::Headers) -> Result<Self::Poll, ConnectionError> {
+        let body = if headers.is_end_stream() {
+            BodyType::Empty
+        } else {
+            BodyType::Stream
+        };
         headers.into_response()
             // TODO: Is this always a protocol error?
             .map_err(|_| ProtocolError.into())
+            .map(move |r| {
+                let (p, _) = r.into_parts();
+                Response::from_parts(p, body)
+            })
     }
 }

src/frame/headers.rs

@@ -1,7 +1,6 @@
+use {hpack, BodyType, HeaderMap};
 use super::StreamId;
-use hpack;
 use frame::{self, Frame, Head, Kind, Error};
-use HeaderMap;
 
 use http::{self, request, response, version, uri, Method, StatusCode, Uri};
 use http::{Request, Response};
@@ -200,14 +199,20 @@ impl Headers {
         self.flags.set_end_stream()
     }
 
-    pub fn into_response(self) -> http::Result<Response<()>> {
+    pub fn into_response(self) -> http::Result<Response<BodyType>> {
         let mut b = Response::builder();
 
         if let Some(status) = self.pseudo.status {
             b.status(status);
         }
 
-        let mut response = try!(b.body(()));
+        let body = if self.is_end_stream() {
+            BodyType::Empty
+        } else {
+            BodyType::Stream
+        };
+
+        let mut response = try!(b.body(body));
         *response.headers_mut() = self.fields;
 
         Ok(response)

src/lib.rs

@@ -34,15 +34,22 @@ mod proto;
 mod frame;
 // pub mod server;
 
+pub use client::Client;
 pub use error::{ConnectionError, Reason};
 pub use frame::StreamId;
+pub use proto::WindowSize;
 
 use bytes::Bytes;
 
 pub type FrameSize = u32;
 // TODO: remove if carllerche/http#90 lands
 pub type HeaderMap = http::HeaderMap<http::header::HeaderValue>;
 
+pub enum BodyType {
+    Empty,
+    Stream,
+}
+
 /// An H2 connection frame
 #[derive(Debug)]
 pub enum Frame<T, B = Bytes> {

src/proto/connection.rs

@@ -127,19 +127,7 @@ impl<T, P, B> Connection<T, P, B>
                     */
                 }
                 Some(Data(frame)) => {
-                    unimplemented!();
-                    /*
-                    trace!("recv DATA; frame={:?}", frame);
-                    try!(self.streams.recv_data(&frame));
-
-                    let frame = Frame::Data {
-                        id: frame.stream_id(),
-                        end_of_stream: frame.is_end_stream(),
-                        data: frame.into_payload(),
-                    };
-
-                    return Ok(Some(frame).into());
-                    */
+                    try!(self.streams.recv_data(frame));
                 }
                 Some(Reset(frame)) => {
                     unimplemented!();

src/proto/streams/recv.rs

@@ -1,4 +1,4 @@
-use {client, frame, ConnectionError};
+use {client, frame, BodyType, ConnectionError};
 use proto::*;
 use super::*;
 
@@ -73,7 +73,7 @@ impl<P, B> Recv<P, B>
     /// Transition the stream state based on receiving headers
     pub fn recv_headers(&mut self,
                         frame: frame::Headers,
-                        stream: &mut store::Ptr<B>)
+                        stream: &mut Stream<B>)
         -> Result<Option<frame::Headers>, ConnectionError>
     {
         stream.state.recv_open(self.init_window_sz, frame.is_end_stream())?;
@@ -98,35 +98,43 @@ impl<P, B> Recv<P, B>
     }
 
     pub fn recv_data(&mut self,
-                     frame: &frame::Data,
+                     frame: frame::Data,
                      stream: &mut Stream<B>)
         -> Result<(), ConnectionError>
     {
         let sz = frame.payload().len();
+        let eos = frame.is_end_stream();
 
         if sz > MAX_WINDOW_SIZE as usize {
             unimplemented!();
         }
 
-        let sz = sz as WindowSize;
+        if sz > 0 {
+            match stream.recv_flow_control() {
+                None => return Err(ProtocolError.into()),
+                Some(flow) => {
+                    let sz = sz as WindowSize;
 
-        match stream.recv_flow_control() {
-            Some(flow) => {
-                // Ensure there's enough capacity on the connection before
-                // acting on the stream.
-                try!(self.flow_control.ensure_window(sz, FlowControlError));
+                    // Ensure there's enough capacity on the connection before
+                    // acting on the stream.
+                    try!(self.flow_control.ensure_window(sz, FlowControlError));
 
-                // Claim the window on the stream
-                try!(flow.claim_window(sz, FlowControlError));
+                    // Claim the window on the stream
+                    try!(flow.claim_window(sz, FlowControlError));
 
-                // Claim the window on the connection.
-                self.flow_control.claim_window(sz, FlowControlError)
-                    .expect("local connection flow control error");
+                    // Claim the window on the connection.
+                    self.flow_control.claim_window(sz, FlowControlError)
+                        .expect("local connection flow control error");
+                }
             }
-            None => return Err(ProtocolError.into()),
         }
 
-        if frame.is_end_stream() {
+        if sz > 0 || eos {
+            stream.pending_recv.push_back(&mut self.buffer, frame.into());
+            stream.notify_recv();
+        }
+
+        if eos {
             try!(stream.state.recv_close());
         }
 
@@ -255,8 +263,9 @@ impl<P, B> Recv<P, B>
 impl<B> Recv<client::Peer, B>
     where B: Buf,
 {
-    pub fn poll_response(&mut self, stream: &mut store::Ptr<B>)
-        -> Poll<Response<()>, ConnectionError> {
+    pub fn poll_response(&mut self, stream: &mut Stream<B>)
+        -> Poll<Response<BodyType>, ConnectionError>
+    {
         // If the buffer is not empty, then the first frame must be a HEADERS
         // frame or the user violated the contract.
         match stream.pending_recv.pop_front(&mut self.buffer) {
@@ -272,4 +281,24 @@ impl<B> Recv<client::Peer, B>
             }
         }
     }
+
+    pub fn poll_data(&mut self, stream: &mut Stream<B>, sz: WindowSize)
+        -> Poll<Option<Bytes>, ConnectionError>
+    {
+        // TODO(ver): split frames into the proper number of bytes, returning unconsumed
+        // bytes onto pending_recv.
+        match stream.pending_recv.pop_front(&mut self.buffer) {
+            Some(Frame::Data(v)) => {
+                unimplemented!()
+            }
+            Some(f) => {
+                stream.pending_recv.push_back(&mut self.buffer, f);
+                Ok(Async::Ready(None))
+            },
+            None => {
+                stream.recv_task = Some(task::current());
+                Ok(Async::NotReady)
+            }
+        }
+    }
 }

src/proto/streams/streams.rs

@@ -1,4 +1,4 @@
-use client;
+use {client, BodyType};
 use proto::*;
 use super::*;
 
@@ -103,7 +103,7 @@ impl<P, B> Streams<P, B>
         Ok(ret)
     }
 
-    pub fn recv_data(&mut self, frame: &frame::Data)
+    pub fn recv_data(&mut self, frame: frame::Data)
         -> Result<(), ConnectionError>
     {
         let id = frame.stream_id();
@@ -308,29 +308,38 @@ impl<B> Streams<client::Peer, B>
 impl<B> StreamRef<client::Peer, B>
     where B: Buf,
 {
-    pub fn poll_response(&mut self) -> Poll<Response<()>, ConnectionError> {
+    pub fn poll_response(&mut self) -> Poll<Response<BodyType>, ConnectionError> {
         let mut me = self.inner.lock().unwrap();
         let me = &mut *me;
 
         let mut stream = me.store.resolve(self.key);
 
         me.actions.recv.poll_response(&mut stream)
     }
+
+    pub fn poll_data(&mut self, sz: WindowSize) -> Poll<Option<Bytes>, ConnectionError> {
+        let mut me = self.inner.lock().unwrap();
+        let me = &mut *me;
+
+        let mut stream = me.store.resolve(self.key);
+
+        me.actions.recv.poll_data(&mut stream, sz)
+    }
 }
 
 impl<P, B> Actions<P, B>
     where P: Peer,
           B: Buf,
 {
     fn dec_num_streams(&mut self, id: StreamId) {
-        if self.is_local_init(id) {
+        if Self::is_local_init(id) {
             self.send.dec_num_streams();
         } else {
             self.recv.dec_num_streams();
         }
     }
 
-    fn is_local_init(&self, id: StreamId) -> bool {
+    fn is_local_init(id: StreamId) -> bool {
         assert!(!id.is_zero());
         P::is_server() == id.is_server_initiated()
     }