feat(http2): add HTTP/2 support for Client and Server

2018-04-13 13:20:47 -07:00
parent fe1578acf6
commit c119097fd0
25 changed files with 2014 additions and 363 deletions
--- a/src/proto/h1/conn.rs
+++ b/src/proto/h1/conn.rs
@@ -129,7 +129,7 @@ where I: AsyncRead + AsyncWrite,
                    let must_error = self.should_error_on_eof();
                    self.state.close_read();
                    self.io.consume_leading_lines();
-                    let was_mid_parse = !self.io.read_buf().is_empty();
+                    let was_mid_parse = e.is_parse() || !self.io.read_buf().is_empty();
                    return if was_mid_parse || must_error {
                        debug!("parse error ({}) with {} bytes", e, self.io.read_buf().len());
                        self.on_parse_error(e)
@@ -566,7 +566,7 @@ where I: AsyncRead + AsyncWrite,
        match self.io.io_mut().shutdown() {
            Ok(Async::NotReady) => Ok(Async::NotReady),
            Ok(Async::Ready(())) => {
-                trace!("shut down IO");
+                trace!("shut down IO complete");
                Ok(Async::Ready(()))
            }
            Err(e) => {
@@ -599,6 +599,12 @@ where I: AsyncRead + AsyncWrite,
            Ok(())
        }
    }
+
+    // Used in h1::dispatch tests
+    #[cfg(test)]
+    pub(super) fn io_mut(&mut self) -> &mut I {
+        self.io.io_mut()
+    }
 }

 impl<I, B: AsRef<[u8]>, T> fmt::Debug for Conn<I, B, T> {
--- a/src/proto/h1/dispatch.rs
+++ b/src/proto/h1/dispatch.rs
@@ -7,8 +7,8 @@ use tokio_service::Service;
 use body::{Body, Payload};
 use proto::{BodyLength, Conn, Http1Transaction, MessageHead, RequestHead, RequestLine, ResponseHead};

-pub(crate) struct Dispatcher<D, Bs, I, B, T> {
-    conn: Conn<I, B, T>,
+pub(crate) struct Dispatcher<D, Bs: Payload, I, T> {
+    conn: Conn<I, Bs::Data, T>,
    dispatch: D,
    body_tx: Option<::body::Sender>,
    body_rx: Option<Bs>,
@@ -31,23 +31,20 @@ pub struct Server<S: Service> {
 }

 pub struct Client<B> {
-    callback: Option<::client::dispatch::Callback<ClientMsg<B>, Response<Body>>>,
+    callback: Option<::client::dispatch::Callback<Request<B>, Response<Body>>>,
    rx: ClientRx<B>,
 }

-pub type ClientMsg<B> = Request<B>;
+type ClientRx<B> = ::client::dispatch::Receiver<Request<B>, Response<Body>>;

-type ClientRx<B> = ::client::dispatch::Receiver<ClientMsg<B>, Response<Body>>;
-
-impl<D, Bs, I, B, T> Dispatcher<D, Bs, I, B, T>
+impl<D, Bs, I, T> Dispatcher<D, Bs, I, T>
 where
    D: Dispatch<PollItem=MessageHead<T::Outgoing>, PollBody=Bs, RecvItem=MessageHead<T::Incoming>>,
    I: AsyncRead + AsyncWrite,
-    B: AsRef<[u8]>,
    T: Http1Transaction,
-    Bs: Payload<Data=B>,
+    Bs: Payload,
 {
-    pub fn new(dispatch: D, conn: Conn<I, B, T>) -> Self {
+    pub fn new(dispatch: D, conn: Conn<I, Bs::Data, T>) -> Self {
        Dispatcher {
            conn: conn,
            dispatch: dispatch,
@@ -286,13 +283,12 @@ where
 }


-impl<D, Bs, I, B, T> Future for Dispatcher<D, Bs, I, B, T>
+impl<D, Bs, I, T> Future for Dispatcher<D, Bs, I, T>
 where
    D: Dispatch<PollItem=MessageHead<T::Outgoing>, PollBody=Bs, RecvItem=MessageHead<T::Incoming>>,
    I: AsyncRead + AsyncWrite,
-    B: AsRef<[u8]>,
    T: Http1Transaction,
-    Bs: Payload<Data=B>,
+    Bs: Payload,
 {
    type Item = ();
    type Error = ::Error;
@@ -493,11 +489,18 @@ mod tests {
    fn client_read_bytes_before_writing_request() {
        let _ = pretty_env_logger::try_init();
        ::futures::lazy(|| {
-            let io = AsyncIo::new_buf(b"HTTP/1.1 200 OK\r\n\r\n".to_vec(), 100);
+            // Block at 0 for now, but we will release this response before
+            // the request is ready to write later...
+            let io = AsyncIo::new_buf(b"HTTP/1.1 200 OK\r\n\r\n".to_vec(), 0);
            let (mut tx, rx) = ::client::dispatch::channel();
            let conn = Conn::<_, ::Chunk, ClientTransaction>::new(io);
            let mut dispatcher = Dispatcher::new(Client::new(rx), conn);

+            // First poll is needed to allow tx to send...
+            assert!(dispatcher.poll().expect("nothing is ready").is_not_ready());
+            // Unblock our IO, which has a response before we've sent request!
+            dispatcher.conn.io_mut().block_in(100);
+
            let res_rx = tx.try_send(::Request::new(::Body::empty())).unwrap();

            let a1 = dispatcher.poll().expect("error should be sent on channel");
@@ -506,13 +509,6 @@ mod tests {
                .expect("callback poll")
                .expect_err("callback response");

-            /*
-            let err = match async {
-                Async::Ready(result) => result.unwrap_err(),
-                Async::Pending => panic!("callback should be ready"),
-            };
-            */
-
            match (err.0.kind(), err.1) {
                (&::error::Kind::Canceled, Some(_)) => (),
                other => panic!("expected Canceled, got {:?}", other),
--- a/src/proto/h1/mod.rs
+++ b/src/proto/h1/mod.rs
@@ -1,6 +1,8 @@
 pub(crate) use self::conn::Conn;
+pub(crate) use self::dispatch::Dispatcher;
 pub use self::decode::Decoder;
 pub use self::encode::{EncodedBuf, Encoder};
+pub use self::io::Cursor; //TODO: move out of h1::io

 mod conn;
 mod date;
--- a/src/proto/h2/client.rs
+++ b/src/proto/h2/client.rs
@@ -0,0 +1,144 @@
+use bytes::IntoBuf;
+use futures::{Async, Future, Poll, Stream};
+use futures::future::{self, Either};
+use futures::sync::oneshot;
+use h2::client::{Builder, Handshake, SendRequest};
+use tokio_io::{AsyncRead, AsyncWrite};
+
+use body::Payload;
+use ::common::{Exec, Never};
+use super::{PipeToSendStream, SendBuf};
+use ::{Body, Request, Response};
+
+type ClientRx<B> = ::client::dispatch::Receiver<Request<B>, Response<Body>>;
+
+pub struct Client<T, B>
+where
+    B: Payload,
+{
+    executor: Exec,
+    rx: ClientRx<B>,
+    state: State<T, SendBuf<B::Data>>,
+}
+
+enum State<T, B> where B: IntoBuf {
+    Handshaking(Handshake<T, B>),
+    Ready(SendRequest<B>, oneshot::Sender<Never>),
+}
+
+impl<T, B> Client<T, B>
+where
+    T: AsyncRead + AsyncWrite + Send + 'static,
+    B: Payload,
+{
+    pub(crate) fn new(io: T, rx: ClientRx<B>, exec: Exec) -> Client<T, B> {
+        let handshake = Builder::new()
+            // we don't expose PUSH promises yet
+            .enable_push(false)
+            .handshake(io);
+
+        Client {
+            executor: exec,
+            rx: rx,
+            state: State::Handshaking(handshake),
+        }
+    }
+}
+
+impl<T, B> Future for Client<T, B>
+where
+    T: AsyncRead + AsyncWrite + Send + 'static,
+    B: Payload + 'static,
+{
+    type Item = ();
+    type Error = ::Error;
+
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+        loop {
+            let next = match self.state {
+                State::Handshaking(ref mut h) => {
+                    let (request_tx, conn) = try_ready!(h.poll().map_err(::Error::new_h2));
+                    // A oneshot channel is used entirely to detect when the
+                    // 'Client' has been dropped. This is to get around a bug
+                    // in h2 where dropping all SendRequests won't notify a
+                    // parked Connection.
+                    let (tx, rx) = oneshot::channel();
+                    let fut = conn
+                        .map_err(|e| debug!("client h2 connection error: {}", e))
+                        .select2(rx)
+                        .then(|res| match res {
+                            Ok(Either::A(((), _))) |
+                            Err(Either::A(((), _))) => {
+                                // conn has finished either way
+                                Either::A(future::ok(()))
+                            },
+                            Err(Either::B((_, conn))) => {
+                                // oneshot has been dropped, hopefully polling
+                                // the connection some more should start shutdown
+                                // and then close
+                                trace!("send_request dropped, starting conn shutdown");
+                                Either::B(conn)
+                            }
+                            Ok(Either::B((never, _))) => match never {},
+                        });
+                    self.executor.execute(fut);
+                    State::Ready(request_tx, tx)
+                },
+                State::Ready(ref mut tx, _) => {
+                    try_ready!(tx.poll_ready().map_err(::Error::new_h2));
+                    match self.rx.poll() {
+                        Ok(Async::Ready(Some((req, mut cb)))) => {
+                            // check that future hasn't been canceled already
+                            if let Async::Ready(()) = cb.poll_cancel().expect("poll_cancel cannot error") {
+                                trace!("request canceled");
+                                continue;
+                            }
+                            let (head, body) = req.into_parts();
+                            let mut req = ::http::Request::from_parts(head, ());
+                            super::strip_connection_headers(req.headers_mut());
+                            let eos = body.is_end_stream();
+                            let (fut, body_tx) = match tx.send_request(req, eos) {
+                                Ok(ok) => ok,
+                                Err(err) => {
+                                    debug!("client send request error: {}", err);
+                                    let _ = cb.send(Err((::Error::new_h2(err), None)));
+                                    continue;
+                                }
+                            };
+                            if !eos {
+                                let pipe = PipeToSendStream::new(body, body_tx);
+                                self.executor.execute(pipe.map_err(|e| debug!("client request body error: {}", e)));
+                            }
+
+                            let fut = fut
+                                .then(move |result| {
+                                    match result {
+                                        Ok(res) => {
+                                            let res = res.map(::Body::h2);
+                                            let _ = cb.send(Ok(res));
+                                        },
+                                        Err(err) => {
+                                            debug!("client response error: {}", err);
+                                            let _ = cb.send(Err((::Error::new_h2(err), None)));
+                                        }
+                                    }
+                                    Ok(())
+                                });
+                            self.executor.execute(fut);
+                            continue;
+                        },
+
+                        Ok(Async::NotReady) => return Ok(Async::NotReady),
+
+                        Ok(Async::Ready(None)) |
+                        Err(_) => {
+                            trace!("client tx dropped");
+                            return Ok(Async::Ready(()));
+                        }
+                    }
+                },
+            };
+            self.state = next;
+        }
+    }
+}
--- a/src/proto/h2/mod.rs
+++ b/src/proto/h2/mod.rs
@@ -1 +1,121 @@
+use bytes::Buf;
+use futures::{Async, Future, Poll};
+use h2::{Reason, SendStream};
+use http::HeaderMap;
+use http::header::{CONNECTION, TRANSFER_ENCODING};

+use ::body::Payload;
+use ::proto::h1::Cursor;
+
+mod client;
+mod server;
+
+pub(crate) use self::client::Client;
+pub(crate) use self::server::Server;
+
+fn strip_connection_headers(headers: &mut HeaderMap) {
+    if headers.remove(TRANSFER_ENCODING).is_some() {
+        trace!("removed illegal Transfer-Encoding header");
+    }
+    if headers.contains_key(CONNECTION) {
+        warn!("Connection header illegal in HTTP/2");
+        //TODO: actually remove it, after checking the value
+        //and removing all related headers
+    }
+}
+
+// body adapters used by both Client and Server
+
+struct PipeToSendStream<S>
+where
+    S: Payload,
+{
+    body_tx: SendStream<SendBuf<S::Data>>,
+    stream: S,
+}
+
+impl<S> PipeToSendStream<S>
+where
+    S: Payload,
+{
+    fn new(stream: S, tx: SendStream<SendBuf<S::Data>>) -> PipeToSendStream<S> {
+        PipeToSendStream {
+            body_tx: tx,
+            stream: stream,
+        }
+    }
+}
+
+impl<S> Future for PipeToSendStream<S>
+where
+    S: Payload,
+{
+    type Item = ();
+    type Error = ::Error;
+
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+        loop {
+            // TODO: make use of flow control on SendStream
+            // If you're looking at this and thinking of trying to fix this TODO,
+            // you may want to look at:
+            // https://docs.rs/h2/0.1.*/h2/struct.SendStream.html
+            //
+            // With that doc open, we'd want to do these things:
+            // - check self.body_tx.capacity() to see if we can send *any* data
+            // - if > 0:
+            // -   poll self.stream
+            // -   reserve chunk.len() more capacity (because its about to be used)?
+            // -   send the chunk
+            // - else:
+            // -   try reserve a smallish amount of capacity
+            // -   call self.body_tx.poll_capacity(), return if NotReady
+            match self.stream.poll_data() {
+                Ok(Async::Ready(Some(chunk))) => {
+                    trace!("send body chunk: {}B", chunk.as_ref().len());
+                    self.body_tx.send_data(SendBuf(Some(Cursor::new(chunk))), false)
+                        .map_err(::Error::new_h2)?;
+                },
+                Ok(Async::Ready(None)) => {
+                    trace!("send body eos");
+                    self.body_tx.send_data(SendBuf(None), true)
+                        .map_err(::Error::new_h2)?;
+                    return Ok(Async::Ready(()));
+                },
+                Ok(Async::NotReady) => return Ok(Async::NotReady),
+                Err(err) => {
+                    let err = ::Error::new_user_body(err);
+                    trace!("send body user stream error: {}", err);
+                    self.body_tx.send_reset(Reason::INTERNAL_ERROR);
+                    return Err(err);
+                }
+            }
+        }
+    }
+}
+
+struct SendBuf<B>(Option<Cursor<B>>);
+
+impl<B: AsRef<[u8]>> Buf for SendBuf<B> {
+    #[inline]
+    fn remaining(&self) -> usize {
+        self.0
+            .as_ref()
+            .map(|b| b.remaining())
+            .unwrap_or(0)
+    }
+
+    #[inline]
+    fn bytes(&self) -> &[u8] {
+        self.0
+            .as_ref()
+            .map(|b| b.bytes())
+            .unwrap_or(&[])
+    }
+
+    #[inline]
+    fn advance(&mut self, cnt: usize) {
+        self.0
+            .as_mut()
+            .map(|b| b.advance(cnt));
+    }
+}
--- a/src/proto/h2/server.rs
+++ b/src/proto/h2/server.rs
@@ -0,0 +1,198 @@
+use futures::{Async, Future, Poll, Stream};
+use h2::Reason;
+use h2::server::{Builder, Connection, Handshake, SendResponse};
+use tokio_io::{AsyncRead, AsyncWrite};
+
+use ::body::Payload;
+use ::common::Exec;
+use ::server::Service;
+use super::{PipeToSendStream, SendBuf};
+
+use ::{Body, Request, Response};
+
+pub(crate) struct Server<T, S, B>
+where
+    S: Service,
+    B: Payload,
+{
+    exec: Exec,
+    service: S,
+    state: State<T, B>,
+}
+
+enum State<T, B>
+where
+    B: Payload,
+{
+    Handshaking(Handshake<T, SendBuf<B::Data>>),
+    Serving(Serving<T, B>),
+}
+
+struct Serving<T, B>
+where
+    B: Payload,
+{
+    conn: Connection<T, SendBuf<B::Data>>,
+}
+
+
+impl<T, S, B> Server<T, S, B>
+where
+    T: AsyncRead + AsyncWrite,
+    S: Service<Request=Request<Body>, Response=Response<B>>,
+    S::Error: Into<Box<::std::error::Error + Send + Sync>>,
+    S::Future: Send + 'static,
+    B: Payload,
+{
+    pub(crate) fn new(io: T, service: S, exec: Exec) -> Server<T, S, B> {
+        let handshake = Builder::new()
+            .handshake(io);
+        Server {
+            exec,
+            state: State::Handshaking(handshake),
+            service,
+        }
+    }
+
+    pub fn graceful_shutdown(&mut self) {
+        unimplemented!("h2 server graceful shutdown");
+    }
+}
+
+impl<T, S, B> Future for Server<T, S, B>
+where
+    T: AsyncRead + AsyncWrite,
+    S: Service<Request=Request<Body>, Response=Response<B>>,
+    S::Error: Into<Box<::std::error::Error + Send + Sync>>,
+    S::Future: Send + 'static,
+    B: Payload,
+{
+    type Item = ();
+    type Error = ::Error;
+
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+        loop {
+            let next = match self.state {
+                State::Handshaking(ref mut h) => {
+                    let conn = try_ready!(h.poll().map_err(::Error::new_h2));
+                    State::Serving(Serving {
+                        conn: conn,
+                    })
+                },
+                State::Serving(ref mut srv) => {
+                    return srv.poll_server(&mut self.service, &self.exec);
+                }
+            };
+            self.state = next;
+        }
+    }
+}
+
+impl<T, B> Serving<T, B>
+where
+    T: AsyncRead + AsyncWrite,
+    B: Payload,
+{
+    fn poll_server<S>(&mut self, service: &mut S, exec: &Exec) -> Poll<(), ::Error>
+    where
+        S: Service<
+            Request=Request<Body>,
+            Response=Response<B>,
+        >,
+        S::Error: Into<Box<::std::error::Error + Send + Sync>>,
+        S::Future: Send + 'static,
+    {
+        while let Some((req, respond)) = try_ready!(self.conn.poll().map_err(::Error::new_h2)) {
+            trace!("incoming request");
+            let req = req.map(::Body::h2);
+            let fut = H2Stream::new(service.call(req), respond);
+            exec.execute(fut);
+        }
+
+        // no more incoming streams...
+        trace!("incoming connection complete");
+        Ok(Async::Ready(()))
+    }
+}
+
+struct H2Stream<F, B>
+where
+    B: Payload,
+{
+    reply: SendResponse<SendBuf<B::Data>>,
+    state: H2StreamState<F, B>,
+}
+
+enum H2StreamState<F, B>
+where
+    B: Payload,
+{
+    Service(F),
+    Body(PipeToSendStream<B>),
+}
+
+impl<F, B> H2Stream<F, B>
+where
+    F: Future<Item=Response<B>>,
+    F::Error: Into<Box<::std::error::Error + Send + Sync>>,
+    B: Payload,
+{
+    fn new(fut: F, respond: SendResponse<SendBuf<B::Data>>) -> H2Stream<F, B> {
+        H2Stream {
+            reply: respond,
+            state: H2StreamState::Service(fut),
+        }
+    }
+
+    fn poll2(&mut self) -> Poll<(), ::Error> {
+        loop {
+            let next = match self.state {
+                H2StreamState::Service(ref mut h) => {
+                    let res = try_ready!(h.poll().map_err(::Error::new_user_service));
+                    let (head, body) = res.into_parts();
+                    let mut res = ::http::Response::from_parts(head, ());
+                    super::strip_connection_headers(res.headers_mut());
+                    macro_rules! reply {
+                        ($eos:expr) => ({
+                            match self.reply.send_response(res, $eos) {
+                                Ok(tx) => tx,
+                                Err(e) => {
+                                    trace!("send response error: {}", e);
+                                    self.reply.send_reset(Reason::INTERNAL_ERROR);
+                                    return Err(::Error::new_h2(e));
+                                }
+                            }
+                        })
+                    }
+                    if !body.is_end_stream() {
+                        let body_tx = reply!(false);
+                        H2StreamState::Body(PipeToSendStream::new(body, body_tx))
+                    } else {
+                        reply!(true);
+                        return Ok(Async::Ready(()));
+                    }
+                },
+                H2StreamState::Body(ref mut pipe) => {
+                    return pipe.poll();
+                }
+            };
+            self.state = next;
+        }
+    }
+}
+
+impl<F, B> Future for H2Stream<F, B>
+where
+    F: Future<Item=Response<B>>,
+    F::Error: Into<Box<::std::error::Error + Send + Sync>>,
+    B: Payload,
+{
+    type Item = ();
+    type Error = ();
+
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+        self.poll2()
+            .map_err(|e| debug!("stream error: {}", e))
+    }
+}
+
--- a/src/proto/mod.rs
+++ b/src/proto/mod.rs
@@ -6,8 +6,8 @@ use headers;

 pub(crate) use self::h1::{dispatch, Conn};

-mod h1;
-//mod h2;
+pub(crate) mod h1;
+pub(crate) mod h2;


 /// An Incoming Message head. Includes request/status line, and headers.