Start hooking up sending data

Author: carllerche

src/client.rs

@@ -160,7 +160,7 @@ impl<B: IntoBuf> Stream<B> {
     pub fn send_data(&mut self, data: B, end_of_stream: bool)
         -> Result<(), ConnectionError>
     {
-        unimplemented!();
+        self.inner.send_data(data.into_buf(), end_of_stream)
     }
 
     /// Send trailers

src/proto/streams/recv.rs

@@ -73,7 +73,7 @@ impl<P, B> Recv<P, B>
         // Increment the number of remote initiated streams
         self.num_streams += 1;
 
-        Ok(Some(Stream::new()))
+        Ok(Some(Stream::new(id)))
     }
 
     /// Transition the stream state based on receiving headers

src/proto/streams/send.rs

@@ -67,7 +67,7 @@ impl<P, B> Send<P, B>
     /// Update state reflecting a new, locally opened stream
     ///
     /// Returns the stream state if successful. `None` if refused
-    pub fn open(&mut self) -> Result<(StreamId, Stream<B>), ConnectionError> {
+    pub fn open(&mut self) -> Result<Stream<B>, ConnectionError> {
         try!(self.ensure_can_open());
 
         if let Some(max) = self.max_streams {
@@ -76,7 +76,7 @@ impl<P, B> Send<P, B>
             }
         }
 
-        let ret = (self.next_stream_id, Stream::new());
+        let ret = Stream::new(self.next_stream_id);
 
         // Increment the number of locally initiated streams
         self.num_streams += 1;
@@ -106,8 +106,8 @@ impl<P, B> Send<P, B>
     }
 
     pub fn send_data(&mut self,
-                     frame: &frame::Data<B>,
-                     stream: &mut Stream<B>)
+                     frame: frame::Data<B>,
+                     stream: &mut store::Ptr<B>)
         -> Result<(), ConnectionError>
     {
         let sz = frame.payload().remaining();
@@ -148,6 +148,8 @@ impl<P, B> Send<P, B>
             try!(stream.state.send_close());
         }
 
+        self.prioritize.queue_frame(frame.into(), stream);
+
         Ok(())
     }
 

src/proto/streams/stream.rs

@@ -2,6 +2,9 @@ use super::*;
 
 #[derive(Debug)]
 pub(super) struct Stream<B> {
+    /// The h2 stream identifier
+    pub id: StreamId,
+
     /// Current state of the stream
     pub state: State,
 
@@ -22,8 +25,9 @@ pub(super) struct Stream<B> {
 }
 
 impl<B> Stream<B> {
-    pub fn new() -> Stream<B> {
+    pub fn new(id: StreamId) -> Stream<B> {
         Stream {
+            id,
             state: State::default(),
             pending_recv: buffer::Deque::new(),
             recv_task: None,

src/proto/streams/streams.rs

@@ -200,6 +200,7 @@ impl<P, B> Streams<P, B>
         */
     }
 
+    /*
     pub fn send_data(&mut self, frame: &frame::Data<B>)
         -> Result<(), ConnectionError>
     {
@@ -222,6 +223,7 @@ impl<P, B> Streams<P, B>
 
         Ok(())
     }
+    */
 
     pub fn poll_window_update(&mut self)
         -> Poll<WindowUpdate, ConnectionError>
@@ -290,13 +292,13 @@ impl<B> Streams<client::Peer, B>
             let me = &mut *me;
 
             // Initialize a new stream. This fails if the connection is at capacity.
-            let (id, mut stream) = me.actions.send.open()?;
+            let mut stream = me.actions.send.open()?;
 
             // Convert the message
             let headers = client::Peer::convert_send_message(
-                id, request, end_of_stream);
+                stream.id, request, end_of_stream);
 
-            let mut stream = me.store.insert(id, stream);
+            let mut stream = me.store.insert(stream.id, stream);
 
             me.actions.send.send_headers(headers, &mut stream)?;
 
@@ -320,6 +322,27 @@ impl<P, B> StreamRef<P, B>
     where P: Peer,
           B: Buf,
 {
+    pub fn send_data(&mut self, data: B, end_of_stream: bool)
+        -> Result<(), ConnectionError>
+    {
+        let mut me = self.inner.lock().unwrap();
+        let me = &mut *me;
+
+        let mut stream = me.store.resolve(self.key);
+
+        // Create the data frame
+        let frame = frame::Data::from_buf(stream.id, data, end_of_stream);
+
+        // Send the data frame
+        me.actions.send.send_data(frame, &mut stream)?;
+
+        if stream.state.is_closed() {
+            me.actions.dec_num_streams(stream.id);
+        }
+
+        Ok(())
+    }
+
     pub fn poll_data(&mut self) -> Poll<Option<Chunk<P, B>>, ConnectionError> {
         let recv = {
             let mut me = self.inner.lock().unwrap();

tests/stream_states.rs

@@ -38,7 +38,6 @@ fn send_recv_headers_only() {
     h2.wait().unwrap();
 }
 
-/*
 #[test]
 fn send_recv_data() {
     let _ = env_logger::init();
@@ -64,14 +63,42 @@ fn send_recv_data() {
         ])
         .build();
 
-    let h2 = client::handshake(mock).wait().expect("handshake");
+    let mut h2 = Client::handshake2(mock)
+        .wait().unwrap();
 
-    // Send the request
-    let mut request = request::Head::default();
-    request.method = method::POST;
-    request.uri = "https://http2.akamai.com/".parse().unwrap();
-    let h2 = h2.send_request(1.into(), request, false).wait().expect("send request");
+    let request = Request::builder()
+        .method(method::POST)
+        .uri("https://http2.akamai.com/")
+        .body(()).unwrap();
+
+    info!("sending request");
+    let mut stream = h2.request(request, false).unwrap();
+
+    // Send the data
+    stream.send_data("hello", true).unwrap();
+
+    // Get the response
+    let resp = h2.run(poll_fn(|| stream.poll_response())).unwrap();
+    assert_eq!(resp.status(), status::OK);
+
+    // Take the body
+    let (_, body) = resp.into_parts();
+
+    // Wait for all the data frames to be received
+    let mut chunks = h2.run(body.collect()).unwrap();
 
+    // Only one chunk since two frames are coalesced.
+    assert_eq!(1, chunks.len());
+
+    let data = chunks[0].pop_bytes().unwrap();
+    assert_eq!(data, &b"world"[..]);
+
+    assert!(chunks[0].pop_bytes().is_none());
+
+    // The H2 connection is closed
+    h2.wait().unwrap();
+
+    /*
     let b = "hello";
 
     // Send the data
@@ -100,8 +127,8 @@ fn send_recv_data() {
     }
 
     assert!(Stream::wait(h2).next().is_none());;
+    */
 }
-*/
 
 #[test]
 fn send_headers_recv_data_single_frame() {
@@ -151,6 +178,8 @@ fn send_headers_recv_data_single_frame() {
     let data = chunks[0].pop_bytes().unwrap();
     assert_eq!(data, &b"world"[..]);
 
+    assert!(chunks[0].pop_bytes().is_none());
+
     // The H2 connection is closed
     h2.wait().unwrap();
 }