feat(http2): add 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)
  }
 }