Wire up Trailers frames

Author: olix0r

src/frame/headers.rs

@@ -198,6 +198,10 @@ impl Headers {
         self.flags.set_end_stream()
     }
 
+    pub fn into_fields(self) -> HeaderMap<HeaderValue> {
+        self.fields
+    }
+
     pub fn into_response(self) -> response::Head {
         let mut response = response::Head::default();
 

src/lib.rs

@@ -42,6 +42,8 @@ use bytes::Bytes;
 
 pub type FrameSize = u32;
 
+pub type HeaderMap = http::HeaderMap<http::header::HeaderValue>;
+
 /// An H2 connection frame
 #[derive(Debug)]
 pub enum Frame<T, B = Bytes> {
@@ -57,7 +59,7 @@ pub enum Frame<T, B = Bytes> {
     },
     Trailers {
         id: StreamId,
-        headers: (),
+        headers: HeaderMap,
     },
     PushPromise {
         id: StreamId,

src/proto/connection.rs

@@ -1,4 +1,4 @@
-use {ConnectionError, Frame, Peer};
+use {ConnectionError, HeaderMap, Frame, Peer};
 use frame::{self, StreamId};
 use client::Client;
 use server::Server;
@@ -110,6 +110,17 @@ impl<T, P, B> Connection<T, P, B>
         })
     }
 
+    pub fn send_trailers(self,
+                     id: StreamId,
+                     headers: HeaderMap)
+        -> sink::Send<Self>
+    {
+        self.send(Frame::Trailers {
+            id,
+            headers,
+        })
+    }
+
     // ===== Private =====
 
     /// Returns `Ready` when the `Connection` is ready to receive a frame from
@@ -165,7 +176,7 @@ impl<T, P, B> Connection<T, P, B>
                 Some(Headers(frame)) => {
                     trace!("recv HEADERS; frame={:?}", frame);
                     // Update stream state while ensuring that the headers frame
-                    // can be received
+                    // can be received.
                     if let Some(frame) = try!(self.streams.recv_headers(frame)) {
                         let frame = Self::convert_poll_message(frame);
                         return Ok(Some(frame).into());
@@ -222,8 +233,10 @@ impl<T, P, B> Connection<T, P, B>
 
     fn convert_poll_message(frame: frame::Headers) -> Frame<P::Poll> {
         if frame.is_trailers() {
-            // TODO: return trailers
-            unimplemented!();
+            Frame::Trailers {
+                id: frame.stream_id(),
+                headers: frame.into_fields()
+            }
         } else {
             Frame::Headers {
                 id: frame.stream_id(),
@@ -338,10 +351,16 @@ impl<T, P, B> Sink for Connection<T, P, B>
 
             Frame::Reset { id, error } => frame::Reset::new(id, error).into(),
 
-            /*
             Frame::Trailers { id, headers } => {
-                unimplemented!();
+                let mut f = frame::Headers::new(id, frame::Pseudo::default(), headers);
+                f.set_end_stream();
+
+                self.streams.send_headers(&f)?;
+
+                frame::Frame::Headers(f)
             }
+
+            /*
             Frame::PushPromise { id, promise } => {
                 unimplemented!();
             }

src/proto/streams.rs

@@ -133,7 +133,7 @@ impl Streams {
         };
 
         if frame.is_trailers() {
-            try!(self.inner.recv_trailers(id, state, frame.is_end_stream()));
+            try!(self.inner.recv_eos(id, state));
         } else {
             try!(self.inner.recv_headers(id, state, frame.is_end_stream()));
         }
@@ -222,7 +222,7 @@ impl Streams {
         };
 
         if frame.is_trailers() {
-            try!(self.inner.send_trailers(id, state, frame.is_end_stream()));
+            try!(self.inner.send_eos(id, state));
         } else {
             try!(self.inner.send_headers(id, state, frame.is_end_stream()));
         }
@@ -326,10 +326,16 @@ impl Inner {
         Ok(())
     }
 
-    fn recv_trailers(&mut self, _id: StreamId, _state: &mut state::Stream, _eos: bool)
+    fn recv_eos(&mut self, id: StreamId, state: &mut state::Stream)
         -> Result<(), ConnectionError>
     {
-        unimplemented!();
+        try!(state.recv_close());
+
+        if state.is_closed() {
+            self.stream_closed(id)
+        }
+
+        Ok(())
     }
 
     fn recv_data(&mut self,
@@ -356,11 +362,7 @@ impl Inner {
         }
 
         if eos {
-            try!(state.recv_close());
-
-            if state.is_closed() {
-                self.stream_closed(id)
-            }
+            self.recv_eos(id, state)?
         }
 
         Ok(())
@@ -396,10 +398,16 @@ impl Inner {
         Ok(())
     }
 
-    fn send_trailers(&mut self, _id: StreamId, _state: &mut state::Stream, _eos: bool)
+    fn send_eos(&mut self, id: StreamId, state: &mut state::Stream)
         -> Result<(), ConnectionError>
     {
-        unimplemented!();
+        try!(state.send_close());
+
+        if state.is_closed() {
+            self.stream_closed(id);
+        }
+
+        Ok(())
     }
 
     fn send_data(&mut self,
@@ -435,11 +443,7 @@ impl Inner {
         }
 
         if eos {
-            try!(state.send_close());
-
-            if state.is_closed() {
-                self.stream_closed(id)
-            }
+            self.send_eos(id, state)?;
         }
 
         Ok(())