Merge pull request #1432 from hyperium/h2

Initial HTTP/2 support for Client and Server

Cargo.toml

@@ -27,6 +27,7 @@ futures-cpupool = "0.1.6"
 futures-timer = "0.1.0"
 http = "0.1.5"
 httparse = "1.0"
+h2 = "0.1.5"
 iovec = "0.1"
 log = "0.4"
 net2 = "0.2.32"
@@ -35,7 +36,7 @@ tokio = "0.1.5"
 tokio-executor = "0.1.0"
 tokio-service = "0.1"
 tokio-io = "0.1"
-want = "0.0.2"
+want = "0.0.3"
 
 [dev-dependencies]
 num_cpus = "1.0"

examples/multi_server.rs

@@ -51,12 +51,12 @@ fn main() {
         println!("Listening on http://{}", srv2.incoming_ref().local_addr());
 
         tokio::spawn(srv1.for_each(move |conn| {
-            tokio::spawn(conn.map(|_| ()).map_err(|err| println!("srv1 error: {:?}", err)));
+            tokio::spawn(conn.map_err(|err| println!("srv1 error: {:?}", err)));
             Ok(())
         }).map_err(|_| ()));
 
         tokio::spawn(srv2.for_each(move |conn| {
-            tokio::spawn(conn.map(|_| ()).map_err(|err| println!("srv2 error: {:?}", err)));
+            tokio::spawn(conn.map_err(|err| println!("srv2 error: {:?}", err)));
             Ok(())
         }).map_err(|_| ()));
 

examples/web_api.rs

@@ -83,7 +83,7 @@ fn main() {
         println!("Listening on http://{} with 1 thread.", serve.incoming_ref().local_addr());
 
         serve.map_err(|_| ()).for_each(move |conn| {
-            tokio::spawn(conn.map(|_| ()).map_err(|err| println!("serve error: {:?}", err)))
+            tokio::spawn(conn.map_err(|err| println!("serve error: {:?}", err)))
         })
     }));
 }

src/body.rs

@@ -5,6 +5,7 @@ use std::fmt;
 use bytes::Bytes;
 use futures::{Async, Future, Poll, Stream};
 use futures::sync::{mpsc, oneshot};
+use h2;
 use http::HeaderMap;
 
 use common::Never;
@@ -13,9 +14,9 @@ use super::Chunk;
 type BodySender = mpsc::Sender<Result<Chunk, ::Error>>;
 
 /// This trait represents a streaming body of a `Request` or `Response`.
-pub trait Payload {
+pub trait Payload: Send + 'static {
     /// A buffer of bytes representing a single chunk of a body.
-    type Data: AsRef<[u8]>;
+    type Data: AsRef<[u8]> + Send;
 
     /// The error type of this stream.
     type Error: Into<Box<::std::error::Error + Send + Sync>>;
@@ -107,6 +108,7 @@ enum Kind {
         _close_tx: oneshot::Sender<()>,
         rx: mpsc::Receiver<Result<Chunk, ::Error>>,
     },
+    H2(h2::RecvStream),
     Wrapped(Box<Stream<Item=Chunk, Error=Box<::std::error::Error + Send + Sync>> + Send>),
     Once(Option<Chunk>),
     Empty,
@@ -219,6 +221,10 @@ impl Body {
         }
     }
 
+    pub(crate) fn h2(recv: h2::RecvStream) -> Self {
+        Body::new(Kind::H2(recv))
+    }
+
     pub(crate) fn delayed_eof(&mut self, fut: DelayEofUntil) {
         self.delayed_eof = Some(DelayEof::NotEof(fut));
     }
@@ -269,6 +275,17 @@ impl Body {
                 Async::Ready(None) => Ok(Async::Ready(None)),
                 Async::NotReady => Ok(Async::NotReady),
             },
+            Kind::H2(ref mut h2) => {
+                h2.poll()
+                    .map(|async| {
+                        async.map(|opt| {
+                            opt.map(|bytes| {
+                                Chunk::h2(bytes, h2.release_capacity())
+                            })
+                        })
+                    })
+                    .map_err(::Error::new_body)
+            },
             Kind::Wrapped(ref mut s) => s.poll().map_err(::Error::new_body),
             Kind::Once(ref mut val) => Ok(Async::Ready(val.take())),
             Kind::Empty => Ok(Async::Ready(None)),
@@ -291,9 +308,17 @@ impl Payload for Body {
         self.poll_eof()
     }
 
+    fn poll_trailers(&mut self) -> Poll<Option<HeaderMap>, Self::Error> {
+        match self.kind {
+            Kind::H2(ref mut h2) => h2.poll_trailers().map_err(::Error::new_h2),
+            _ => Ok(Async::Ready(None)),
+        }
+    }
+
     fn is_end_stream(&self) -> bool {
         match self.kind {
             Kind::Chan { .. } => false,
+            Kind::H2(..) => false,
             Kind::Wrapped(..) => false,
             Kind::Once(ref val) => val.is_none(),
             Kind::Empty => true
@@ -303,6 +328,7 @@ impl Payload for Body {
     fn content_length(&self) -> Option<u64> {
         match self.kind {
             Kind::Chan { .. } => None,
+            Kind::H2(..) => None,
             Kind::Wrapped(..) => None,
             Kind::Once(Some(ref val)) => Some(val.len() as u64),
             Kind::Once(None) => None,

src/chunk.rs

@@ -1,12 +1,38 @@
 use std::fmt;
 
 use bytes::Bytes;
+use h2::ReleaseCapacity;
 
 /// A piece of a message body.
 pub struct Chunk(Inner);
 
-enum Inner {
-    Shared(Bytes),
+struct Inner {
+    bytes: Bytes,
+    _flow_control: Option<AutoRelease>,
+}
+
+struct AutoRelease {
+    cap: usize,
+    release: ReleaseCapacity,
+}
+
+impl Drop for AutoRelease {
+    fn drop(&mut self) {
+        let _ = self.release.release_capacity(self.cap);
+    }
+}
+
+impl Chunk {
+    pub(crate) fn h2(bytes: Bytes, rel_cap: &ReleaseCapacity) -> Chunk {
+        let cap = bytes.len();
+        Chunk(Inner {
+            bytes: bytes,
+            _flow_control: Some(AutoRelease {
+                cap: cap,
+                release: rel_cap.clone(),
+            }),
+        })
+    }
 }
 
 impl From<Vec<u8>> for Chunk {
@@ -39,17 +65,18 @@ impl From<&'static str> for Chunk {
 
 impl From<Bytes> for Chunk {
     #[inline]
-    fn from(mem: Bytes) -> Chunk {
-        Chunk(Inner::Shared(mem))
+    fn from(bytes: Bytes) -> Chunk {
+        Chunk(Inner {
+            bytes: bytes,
+            _flow_control: None,
+        })
     }
 }
 
 impl From<Chunk> for Bytes {
     #[inline]
     fn from(chunk: Chunk) -> Bytes {
-        match chunk.0 {
-            Inner::Shared(bytes) => bytes,
-        }
+        chunk.0.bytes
     }
 }
 
@@ -65,9 +92,7 @@ impl ::std::ops::Deref for Chunk {
 impl AsRef<[u8]> for Chunk {
     #[inline]
     fn as_ref(&self) -> &[u8] {
-        match self.0 {
-            Inner::Shared(ref slice) => slice,
-        }
+        &self.0.bytes
     }
 }
 
@@ -81,7 +106,7 @@ impl fmt::Debug for Chunk {
 impl Default for Chunk {
     #[inline]
     fn default() -> Chunk {
-        Chunk(Inner::Shared(Bytes::new()))
+        Chunk::from(Bytes::new())
     }
 }
 
@@ -91,17 +116,13 @@ impl IntoIterator for Chunk {
 
     #[inline]
     fn into_iter(self) -> Self::IntoIter {
-        match self.0 {
-            Inner::Shared(bytes) => bytes.into_iter(),
-        }
+        self.0.bytes.into_iter()
     }
 }
 
 impl Extend<u8> for Chunk {
     #[inline]
     fn extend<T>(&mut self, iter: T) where T: IntoIterator<Item=u8> {
-        match self.0 {
-            Inner::Shared(ref mut bytes) => bytes.extend(iter)
-        }
+        self.0.bytes.extend(iter)
     }
 }