Add tower-h2 and more macro

Cargo.toml

@@ -1,5 +1,6 @@
 [workspace]
 members = [
     "tonic",
-    "tonic-macros"
+    "tonic-macros",
+    "tower-h2"
 ]

tonic-macros/Cargo.toml

@@ -8,7 +8,6 @@ edition = "2018"
 proc-macro = true
 
 [dependencies]
-tonic = { path = "../tonic" }
 syn = { version = "0.15", features = ["full"] }
 quote = "0.6"
 proc-macro2 = "0.4"
@@ -17,3 +16,4 @@ tower-service = { git = "https://github.com/tower-rs/tower", branch = "std-futur
 
 [dev-dependencies]
 tokio = "=0.2.0-alpha.1"
+tonic = { path = "../tonic" }

tonic-macros/src/lib.rs

@@ -7,7 +7,7 @@ use quote::quote;
 use syn::{ImplItem, ImplItemMethod, ItemImpl, Type};
 
 #[proc_macro_attribute]
-pub fn grpc(attr: TokenStream, item: TokenStream) -> TokenStream {
+pub fn server(attr: TokenStream, item: TokenStream) -> TokenStream {
     let service = load_service(attr);
     let mut original = item.clone();
     let ItemImpl { self_ty, items, .. } = syn::parse_macro_input!(item as ItemImpl);
@@ -36,33 +36,6 @@ pub fn grpc(attr: TokenStream, item: TokenStream) -> TokenStream {
         }
     }
 
-    // let ts = quote! {
-    //     impl<'a> tower_service::Service<tonic::Request<()>> for #s {
-    //         type Response = tonic::Response<()>;
-    //         type Error = tonic::Status;
-    //         type Future = tonic::ResponseFuture<'a, Self::Response, Self::Error>;
-
-    //         fn poll_ready(&mut self, _cx: &mut std::task::Context<'_>) -> std::task::Poll<Result<(), Self::Error>> {
-    //             std::task::Poll::Ready(Ok(()))
-    //         }
-
-    //         fn call(&mut self, request: tonic::Request<()>) -> Self::Future {
-    //             Box::pin(self.#m_ident(request))
-    //         }
-    //     }
-    // };
-
-    // let ts = quote! {
-    //     impl tonic::GrpcInnerService<tonic::Request<()>> for #s {
-    //         type Response = tonic::Response<()>;
-
-    //         fn call<'a>(&'a mut self, request: tonic::Request<()>) -> tonic::ResponseFuture<'a, Self::Response>
-    //             where Self: 'a {
-    //             Box::pin(self.#m_ident(request))
-    //         }
-    //     }
-    // };
-
     let ts = quote! {
         pub struct GrpcServer {
             inner: std::sync::Arc<#s>,
@@ -88,7 +61,6 @@ pub fn grpc(attr: TokenStream, item: TokenStream) -> TokenStream {
                 Box::pin(async move {
                     inner.#m_ident(request).await
                 })
-                //self.#m_ident(request)
             }
         }
     };

tonic-macros/tests/server.rs

@@ -3,7 +3,6 @@
 use std::time::Duration;
 use tokio::timer::Delay;
 use tonic::{Request, Response, Status};
-use tonic_macros::grpc;
 
 // #[derive(Debug)]
 // struct HelloRequest;
@@ -15,7 +14,7 @@ struct MyGreeter {
     data: String,
 }
 
-#[grpc(service = "proto/helloworld.proto")]
+#[tonic::server(service = "proto/helloworld.proto")]
 impl MyGreeter {
     pub async fn say_hello(&self, request: Request<()>) -> Result<Response<()>, Status> {
         println!("Got a request: {:?}", request);

tonic/Cargo.toml

@@ -8,3 +8,4 @@ edition = "2018"
 
 [dependencies]
 tower-grpc = { git = "https://github.com/tower-rs/tower-grpc", branch = "std-future" }
+tonic-macros = { path = "../tonic-macros" }

tonic/src/lib.rs

@@ -1,5 +1,7 @@
 pub use tower_grpc::*;
 
+pub use tonic_macros::server;
+
 use std::future::Future;
 use std::pin::Pin;
 use std::sync::Arc;

tower-h2/Cargo.toml

@@ -0,0 +1,23 @@
+[package]
+name = "tower-h2"
+version = "0.1.0"
+authors = ["Lucio Franco <luciofranco14@gmail.com>"]
+edition = "2018"
+
+[dependencies]
+futures-core-preview = "=0.3.0-alpha.17"
+futures-util-preview = "=0.3.0-alpha.17"
+bytes = "0.4"
+tokio-io = "0.2.0-alpha.1"
+tokio-executor = "0.2.0-alpha.1"
+tower-service = { git = "http://github.com/tower-rs/tower", branch = "std-future" }
+tower-util = { git = "http://github.com/tower-rs/tower", branch = "std-future" }
+h2 = { git = "https://github.com/LucioFranco/h2", branch = "lucio/tower-h2-hack" }
+http = "0.1"
+http-body = { git =  "https://github.com/hyperium/http-body", branch = "std-future" }
+log = "0.4"
+
+[dev-dependencies]
+tokio = "=0.2.0-alpha.1"
+tower-util = { git = "http://github.com/tower-rs/tower", branch = "std-future" }
+tokio-buf = "=0.2.0-alpha.1"

tower-h2/examples/client.rs

@@ -0,0 +1,49 @@
+#![feature(async_await)]
+
+use http::Request;
+use std::task::{Context, Poll};
+use tokio::net::TcpStream;
+use tokio_buf::BufStream;
+use tower_h2::Connection;
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let addr = "[::1]:8888".parse()?;
+    let io = TcpStream::connect(&addr).await?;
+
+    let mut svc = Connection::handshake(io).await?;
+
+    let req = Request::get(format!("http://{}", addr)).body(Body::from(Vec::new()))?;
+    let res = svc.send(req).await?;
+
+    println!("RESPONSE={:?}", res);
+
+    Ok(())
+}
+
+#[derive(Debug, Default, Clone)]
+struct Body(Vec<u8>);
+
+impl From<Vec<u8>> for Body {
+    fn from(t: Vec<u8>) -> Self {
+        Body(t)
+    }
+}
+
+impl BufStream for Body {
+    type Item = std::io::Cursor<Vec<u8>>;
+    type Error = std::io::Error;
+
+    fn poll_buf(&mut self, _cx: &mut Context<'_>) -> Poll<Option<Result<Self::Item, Self::Error>>> {
+        if self.0.is_empty() {
+            return None.into();
+        }
+
+        use std::{io, mem};
+
+        let bytes = mem::replace(&mut self.0, Default::default());
+        let buf = io::Cursor::new(bytes);
+
+        Some(Ok(buf)).into()
+    }
+}

tower-h2/examples/server.rs

@@ -0,0 +1,103 @@
+#![feature(async_await)]
+
+use futures_util::future;
+use http::{Request, Response};
+use std::task::{Context, Poll};
+use tokio_buf::BufStream;
+use tower_h2::{RecvBody, Server};
+use tower_service::Service;
+use tokio::net::TcpListener;
+
+const ROOT: &'static str = "/";
+
+#[derive(Debug)]
+pub struct Svc;
+
+impl Service<Request<RecvBody>> for Svc {
+    type Response = Response<Body>;
+    type Error = h2::Error;
+    type Future = future::Ready<Result<Self::Response, Self::Error>>;
+
+    fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        Ok(()).into()
+    }
+
+    fn call(&mut self, req: Request<RecvBody>) -> Self::Future {
+        let mut rsp = Response::builder();
+        rsp.version(http::Version::HTTP_2);
+
+        let uri = req.uri();
+        if uri.path() != ROOT {
+            let body = Body::from(Vec::new());
+            let rsp = rsp.status(404).body(body).unwrap();
+            return future::ok(rsp);
+        }
+
+        let body = Body::from(Vec::from(&b"heyo!"[..]));
+        let rsp = rsp.status(200).body(body).unwrap();
+        future::ok(rsp)
+    }
+}
+
+pub struct MakeSvc;
+
+impl Service<()> for MakeSvc {
+    type Response = Svc;
+    type Error = std::io::Error;
+    type Future = future::Ready<Result<Self::Response, Self::Error>>;
+
+    fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        Ok(()).into()
+    }
+
+    fn call(&mut self, _: ()) -> Self::Future {
+        future::ok(Svc)
+    }
+}
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let addr = "[::1]:8888".parse().unwrap();
+    let mut bind = TcpListener::bind(&addr)?;
+
+    let mut server = Server::new(MakeSvc, Default::default());
+
+    while let Ok((sock, _addr)) = bind.accept().await {
+        if let Err(e) = sock.set_nodelay(true) {
+            return Err(e.into());
+        }
+
+        if let Err(e) = server.serve(sock).await {
+            println!("H2 ERROR: {}", e);
+        }
+    }
+
+    Ok(())
+}
+
+#[derive(Debug, Default, Clone)]
+pub struct Body(Vec<u8>);
+
+impl From<Vec<u8>> for Body {
+    fn from(t: Vec<u8>) -> Self {
+        Body(t)
+    }
+}
+
+impl BufStream for Body {
+    type Item = std::io::Cursor<Vec<u8>>;
+    type Error = std::io::Error;
+
+    fn poll_buf(&mut self, _cx: &mut Context<'_>) -> Poll<Option<Result<Self::Item, Self::Error>>> {
+        if self.0.is_empty() {
+            return None.into();
+        }
+
+        use std::{io, mem};
+
+        let bytes = mem::replace(&mut self.0, Default::default());
+        let buf = io::Cursor::new(bytes);
+
+        Some(Ok(buf)).into()
+    }
+}

tower-h2/src/buf.rs

@@ -0,0 +1,38 @@
+use bytes::Buf;
+
+pub struct SendBuf<T> {
+    inner: Option<T>,
+}
+
+impl<T: Buf> SendBuf<T> {
+    pub fn new(buf: T) -> SendBuf<T> {
+        SendBuf { inner: Some(buf) }
+    }
+
+    pub fn none() -> SendBuf<T> {
+        SendBuf { inner: None }
+    }
+}
+
+impl<T: Buf> Buf for SendBuf<T> {
+    fn remaining(&self) -> usize {
+        match self.inner {
+            Some(ref v) => v.remaining(),
+            None => 0,
+        }
+    }
+
+    fn bytes(&self) -> &[u8] {
+        match self.inner {
+            Some(ref v) => v.bytes(),
+            None => &[],
+        }
+    }
+
+    fn advance(&mut self, cnt: usize) {
+        match self.inner {
+            Some(ref mut v) => v.advance(cnt),
+            None => {}
+        }
+    }
+}

tower-h2/src/client.rs

@@ -0,0 +1,81 @@
+use crate::{buf::SendBuf, flush::Flush, recv_body::RecvBody};
+use futures_util::{future, FutureExt, TryFutureExt};
+use h2::{client::SendRequest, RecvStream};
+use http::{Request, Response};
+use http_body::Body;
+use std::future::Future;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+use tokio_io::{AsyncRead, AsyncWrite};
+use tower_service::Service;
+
+type BoxFuture<T> = Pin<Box<dyn Future<Output = T> + Send + 'static>>;
+
+pub struct Connection<B>
+where
+    B: Body + Unpin,
+    B::Data: Unpin,
+{
+    client: SendRequest<SendBuf<B::Data>>,
+}
+
+impl<B> Connection<B>
+where
+    B: Body + Send + Unpin + 'static,
+    B::Data: Send + Unpin + 'static,
+    B::Error: Into<Box<dyn std::error::Error>>,
+{
+    pub async fn handshake<T>(io: T) -> Result<Connection<B>, h2::Error>
+    where
+        T: AsyncRead + AsyncWrite + Send + Unpin + 'static,
+    {
+        let builder = h2::client::Builder::new();
+        let (client, conn) = builder.handshake(io).await?;
+        tokio_executor::spawn(conn.map_err(|e| println!("ERROR={}", e)).map(drop));
+        Ok(Connection { client })
+    }
+
+    pub async fn send(&mut self, request: Request<B>) -> Result<Response<RecvBody>, h2::Error> {
+        future::poll_fn(|cx| self.poll_ready(cx)).await?;
+
+        self.call(request).await
+    }
+}
+
+impl<B> Service<Request<B>> for Connection<B>
+where
+    B: Body + Send + Unpin + 'static,
+    B::Data: Send + Unpin + 'static,
+    B::Error: Into<Box<dyn std::error::Error>>,
+{
+    type Response = Response<RecvBody>;
+    type Error = h2::Error;
+    type Future = BoxFuture<Result<Self::Response, Self::Error>>;
+
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.client.poll_ready(cx)
+    }
+
+    fn call(&mut self, request: Request<B>) -> Self::Future {
+        let (parts, body) = request.into_parts();
+        let request = Request::from_parts(parts, ());
+
+        let eos = body.is_end_stream();
+
+        let res = self.client.send_request(request, eos);
+
+        let (response, send_body) = match res {
+            Ok(success) => success,
+            Err(e) => {
+                return Box::pin(future::err(e));
+            }
+        };
+
+        if !eos {
+            let flush = Flush::new(body, send_body);
+            tokio_executor::spawn(flush.map(drop));
+        }
+
+        Box::pin(response.map_ok(|r| r.map(RecvBody::new)))
+    }
+}

tower-h2/src/error.rs

@@ -0,0 +1,12 @@
+pub(crate) fn reason_from_dyn_error(err: &(dyn std::error::Error + 'static)) -> h2::Reason {
+    let mut cause = Some(err);
+    while let Some(err) = cause {
+        if let Some(h2_err) = err.downcast_ref::<h2::Error>() {
+            return h2_err.reason().unwrap_or(h2::Reason::INTERNAL_ERROR);
+        }
+        cause = err.source();
+    }
+
+    // unknown error
+    h2::Reason::INTERNAL_ERROR
+}

tower-h2/src/flush.rs

@@ -0,0 +1,195 @@
+use crate::buf::SendBuf;
+use futures_util::ready;
+use h2::{self, SendStream};
+use http::HeaderMap;
+use http_body::Body;
+use std::future::Future;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+/// Flush a body to the HTTP/2.0 send stream
+pub(crate) struct Flush<S>
+where
+    S: Body,
+{
+    h2: SendStream<SendBuf<S::Data>>,
+    body: S,
+    state: FlushState,
+}
+
+enum FlushState {
+    Data,
+    Trailers,
+    Done,
+}
+
+enum DataOrTrailers<B> {
+    Data(B),
+    Trailers(HeaderMap),
+}
+
+// ===== impl Flush =====
+
+impl<S> Flush<S>
+where
+    S: Body,
+    S::Error: Into<Box<dyn std::error::Error>>,
+{
+    pub fn new(src: S, dst: SendStream<SendBuf<S::Data>>) -> Self {
+        Flush {
+            h2: dst,
+            body: src,
+            state: FlushState::Data,
+        }
+    }
+
+    /// Try to flush the body.
+    fn poll_complete(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), h2::Error>> {
+        use self::DataOrTrailers::*;
+
+        loop {
+            match ready!(self.poll_body(cx)) {
+                Some(Ok(Data(buf))) => {
+                    let eos = self.body.is_end_stream();
+
+                    self.h2.send_data(SendBuf::new(buf), eos)?;
+
+                    if eos {
+                        self.state = FlushState::Done;
+                        return Ok(()).into();
+                    }
+                }
+                Some(Ok(Trailers(trailers))) => {
+                    self.h2.send_trailers(trailers)?;
+                    return Ok(()).into();
+                }
+                Some(Err(e)) => panic!("error {:?}", e),
+                None => {
+                    // If this is hit, then an EOS was not reached via the other
+                    // paths. So, we must send an empty data frame with EOS.
+                    self.h2.send_data(SendBuf::none(), true)?;
+
+                    return Ok(()).into();
+                }
+            }
+        }
+    }
+
+    /// Get the next message to write, either a data frame or trailers.
+    fn poll_body(
+        &mut self,
+        cx: &mut Context<'_>,
+    ) -> Poll<Option<Result<DataOrTrailers<S::Data>, h2::Error>>> {
+        loop {
+            match self.state {
+                FlushState::Data => {
+                    // Before trying to poll the next chunk, we have to see if
+                    // the h2 connection has capacity. We do this by requesting
+                    // a single byte (since we don't know how big the next chunk
+                    // will be.
+                    self.h2.reserve_capacity(1);
+
+                    if self.h2.capacity() == 0 {
+                        // TODO: The loop should not be needed once
+                        // carllerche/h2#270 is fixed.
+                        loop {
+                            match ready!(self.h2.poll_capacity(cx)) {
+                                Some(Ok(0)) => {}
+                                Some(Ok(_)) => break,
+                                Some(Err(e)) => return panic!("error {:?}", e),
+                                None => {
+                                    debug!("connection closed early");
+                                    // The error shouldn't really matter at this
+                                    // point as the peer has disconnected, the
+                                    // error will be discarded anyway.
+                                    return Some(Err(h2::Reason::INTERNAL_ERROR.into())).into();
+                                }
+                            }
+                        }
+                    } else {
+                        // If there was capacity already assigned, then the
+                        // stream state wasn't polled, but we should fail out
+                        // if the stream has been reset, so we poll for that.
+                        match self.h2.poll_reset(cx) {
+                            Poll::Ready(Ok(reason)) => {
+                                debug!("stream received RST_STREAM while flushing: {:?}", reason,);
+                                return Some(Err(reason.into())).into();
+                            }
+                            Poll::Ready(Err(e)) => return Some(Err(e)).into(),
+                            Poll::Pending => {
+                                // Stream hasn't been reset, so we can try
+                                // to send data below. This task has been
+                                // registered in case data isn't ready
+                                // before we get a RST_STREAM.
+                            }
+                        }
+                    }
+
+                    let item = match ready!(self.body.poll_data(cx)) {
+                        Some(Ok(d)) => Some(d),
+                        Some(Err(err)) => {
+                            let err = err.into();
+                            debug!("user body error from poll_buf: {}", err);
+                            let reason = crate::error::reason_from_dyn_error(&*err);
+                            self.h2.send_reset(reason);
+                            return Some(Err(reason.into())).into();
+                        }
+                        None => None,
+                    };
+
+                    if let Some(data) = item {
+                        return Some(Ok(DataOrTrailers::Data(data))).into();
+                    } else {
+                        // Release all capacity back to the connection
+                        self.h2.reserve_capacity(0);
+                        self.state = FlushState::Trailers;
+                    }
+                }
+                FlushState::Trailers => {
+                    match self.h2.poll_reset(cx) {
+                        Poll::Ready(Ok(reason)) => {
+                            debug!(
+                                "stream received RST_STREAM while flushing trailers: {:?}",
+                                reason,
+                            );
+                            return Some(Err(reason.into())).into();
+                        }
+                        Poll::Ready(Err(e)) => return Some(Err(e)).into(),
+                        Poll::Pending => {
+                            // Stream hasn't been reset, so we can try
+                            // to send data below. This task has been
+                            // registered in case data isn't ready
+                            // before we get a RST_STREAM.
+                        }
+                    }
+                    let trailers = ready!(self.body.poll_trailers(cx).map_err(|err| {
+                        let err = err.into();
+                        debug!("user body error from poll_trailers: {}", err);
+                        let reason = crate::error::reason_from_dyn_error(&*err);
+                        self.h2.send_reset(reason);
+                        reason
+                    }))?;
+                    self.state = FlushState::Done;
+                    if let Some(trailers) = trailers {
+                        return Some(Ok(DataOrTrailers::Trailers(trailers))).into();
+                    }
+                }
+                FlushState::Done => return None.into(),
+            }
+        }
+    }
+}
+
+impl<S> Future for Flush<S>
+where
+    S: Body + Unpin,
+    S::Error: Into<Box<dyn std::error::Error>>,
+{
+    type Output = Result<(), ()>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        Pin::new(&mut self)
+            .poll_complete(cx)
+            .map_err(|err| warn!("error flushing stream: {:?}", err))
+    }
+}

tower-h2/src/lib.rs

@@ -0,0 +1,15 @@
+#![feature(async_await)]
+
+#[macro_use]
+extern crate log;
+
+mod buf;
+mod client;
+mod error;
+mod flush;
+mod recv_body;
+mod server;
+
+pub use client::Connection;
+pub use recv_body::RecvBody;
+pub use server::Server;

tower-h2/src/recv_body.rs

@@ -0,0 +1,95 @@
+use bytes::{Buf, Bytes, BytesMut};
+use futures_core::Stream;
+use futures_util::TryStreamExt;
+use http_body::Body;
+use std::task::{Context, Poll};
+
+/// Allows a stream to be read from the remote.
+#[derive(Debug)]
+pub struct RecvBody {
+    inner: h2::RecvStream,
+}
+
+#[derive(Debug)]
+pub struct Data {
+    bytes: Bytes,
+}
+
+// ===== impl RecvBody =====
+
+impl RecvBody {
+    /// Return a new `RecvBody`.
+    pub(crate) fn new(inner: h2::RecvStream) -> Self {
+        RecvBody { inner }
+    }
+
+    /// Returns the stream ID of the received stream, or `None` if this body
+    /// does not correspond to a stream.
+    pub fn stream_id(&self) -> h2::StreamId {
+        self.inner.stream_id()
+    }
+}
+
+impl Body for RecvBody {
+    type Data = Data;
+    type Error = h2::Error;
+
+    fn is_end_stream(&self) -> bool {
+        self.inner.is_end_stream()
+    }
+
+    fn poll_data(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<Self::Data, h2::Error>>> {
+        let data = match futures_util::ready!(self.inner.try_poll_next_unpin(cx)) {
+            Some(Ok(bytes)) => {
+                self.inner
+                    .release_capacity()
+                    .release_capacity(bytes.len())
+                    .expect("flow control error");
+                Data { bytes }
+            }
+            Some(Err(e)) => return Some(Err(e)).into(),
+            None => return None.into(),
+        };
+
+        Some(Ok(data)).into()
+    }
+
+    fn poll_trailers(
+        &mut self,
+        cx: &mut Context<'_>,
+    ) -> Poll<Result<Option<http::HeaderMap>, h2::Error>> {
+        match futures_util::ready!(self.inner.poll_trailers(cx)) {
+            Some(Ok(t)) => Ok(Some(t)).into(),
+            Some(Err(e)) => Err(e).into(),
+            None => Ok(None).into(),
+        }
+    }
+}
+
+// ===== impl Data =====
+
+impl Buf for Data {
+    fn remaining(&self) -> usize {
+        self.bytes.len()
+    }
+
+    fn bytes(&self) -> &[u8] {
+        self.bytes.as_ref()
+    }
+
+    fn advance(&mut self, cnt: usize) {
+        self.bytes.advance(cnt);
+    }
+}
+
+impl From<Data> for Bytes {
+    fn from(src: Data) -> Self {
+        src.bytes
+    }
+}
+
+impl From<Data> for BytesMut {
+    fn from(src: Data) -> Self {
+        src.bytes.into()
+    }
+}

tower-h2/src/server.rs

@@ -0,0 +1,108 @@
+use crate::{buf::SendBuf, flush::Flush, recv_body::RecvBody};
+use futures_util::{future, StreamExt};
+use http::{Request, Response};
+use http_body::Body;
+use std::marker::PhantomData;
+use tokio_io::{AsyncRead, AsyncWrite};
+use tower_service::Service;
+use tower_util::MakeService;
+
+pub struct Server<M, B>
+where
+    M: MakeService<(), Request<RecvBody>>,
+    B: Body,
+{
+    maker: M,
+    builder: h2::server::Builder,
+    _pd: PhantomData<B>,
+}
+
+impl<M, B> Server<M, B>
+where
+    M: MakeService<(), Request<RecvBody>, Response = Response<B>>,
+    M::MakeError: Into<Box<dyn std::error::Error>>,
+    M::Error: Into<Box<dyn std::error::Error>>,
+    B: Body + Send + Unpin + 'static,
+    B::Data: Send + Unpin,
+    B::Error: Into<Box<dyn std::error::Error>>,
+{
+    pub fn new(maker: M, builder: h2::server::Builder) -> Self {
+        Self {
+            maker,
+            builder,
+            _pd: PhantomData
+        }
+    }
+
+    pub async fn serve<I>(&mut self, io: I) -> Result<(), h2::Error>
+    where
+        I: AsyncRead + AsyncWrite + Unpin,
+    {
+        future::poll_fn(|cx| self.maker.poll_ready(cx))
+            .await
+            .map_err(Into::into)
+            .unwrap();
+        let mut service = self
+            .maker
+            .make_service(())
+            .await
+            .map_err(Into::into)
+            .unwrap();
+
+        let mut connection: h2::server::Connection<I, SendBuf<B::Data>> =
+            self.builder.handshake(io).await?;
+
+        // TODO: do we want to spawn the connectioons o it can poll_close?
+
+        while let Some(request) = connection.next().await {
+            match request {
+                Ok((request, send_response)) => {
+                    let request = request.map(RecvBody::new);
+
+                    future::poll_fn(|cx| service.poll_ready(cx))
+                        .await
+                        .map_err(Into::into)
+                        .unwrap();
+
+                    // TODO: on error send reset
+                    let response = service.call(request).await.map_err(Into::into).unwrap();
+
+                    let fut = handle_request(response, send_response);
+                    tokio_executor::spawn(fut);
+                }
+                Err(e) => return Err(e),
+            }
+        }
+
+        Ok(())
+    }
+}
+
+pub async fn handle_request<B>(
+        response: Response<B>,
+        mut send_response: h2::server::SendResponse<SendBuf<B::Data>>,
+    ) where
+        B: Body + Send + Unpin + 'static,
+        B::Data: Unpin,
+        B::Error: Into<Box<dyn std::error::Error>>,
+    {
+        let (parts, body) = response.into_parts();
+
+        // Check if the response is imemdiately an end-of-stream.
+        let eos = body.is_end_stream();
+
+        let response = Response::from_parts(parts, ());
+
+        match send_response.send_response(response, eos) {
+            Ok(sr) => {
+                if eos {
+                    return;
+                }
+
+                Flush::new(body, sr).await.unwrap();
+            }
+            Err(e) => {
+                println!("h2 server ERROR={}", e);
+            }
+        }
+    }