From 404f7d8931b5647ff6632c4fc8b517e90eb1c460 Mon Sep 17 00:00:00 2001 From: Lucio Franco Date: Sun, 11 Aug 2019 13:48:13 -0400 Subject: [PATCH] Add tower-h2 and more macro --- Cargo.toml | 3 +- tonic-macros/Cargo.toml | 2 +- tonic-macros/src/lib.rs | 30 +--- tonic-macros/tests/{grpc.rs => server.rs} | 3 +- tonic/Cargo.toml | 1 + tonic/src/lib.rs | 2 + tower-h2/Cargo.toml | 23 +++ tower-h2/examples/client.rs | 49 ++++++ tower-h2/examples/server.rs | 103 ++++++++++++ tower-h2/src/buf.rs | 38 +++++ tower-h2/src/client.rs | 81 +++++++++ tower-h2/src/error.rs | 12 ++ tower-h2/src/flush.rs | 195 ++++++++++++++++++++++ tower-h2/src/lib.rs | 15 ++ tower-h2/src/recv_body.rs | 95 +++++++++++ tower-h2/src/server.rs | 108 ++++++++++++ 16 files changed, 727 insertions(+), 33 deletions(-) rename tonic-macros/tests/{grpc.rs => server.rs} (92%) create mode 100644 tower-h2/Cargo.toml create mode 100644 tower-h2/examples/client.rs create mode 100644 tower-h2/examples/server.rs create mode 100644 tower-h2/src/buf.rs create mode 100644 tower-h2/src/client.rs create mode 100644 tower-h2/src/error.rs create mode 100644 tower-h2/src/flush.rs create mode 100644 tower-h2/src/lib.rs create mode 100644 tower-h2/src/recv_body.rs create mode 100644 tower-h2/src/server.rs diff --git a/Cargo.toml b/Cargo.toml index 25c2376..c76403a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,5 +1,6 @@ [workspace] members = [ "tonic", - "tonic-macros" + "tonic-macros", + "tower-h2" ] diff --git a/tonic-macros/Cargo.toml b/tonic-macros/Cargo.toml index faba5ab..45285d8 100644 --- a/tonic-macros/Cargo.toml +++ b/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" } diff --git a/tonic-macros/src/lib.rs b/tonic-macros/src/lib.rs index 7a01aa7..fef66d9 100644 --- a/tonic-macros/src/lib.rs +++ b/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> 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> { - // 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> 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) } } }; diff --git a/tonic-macros/tests/grpc.rs b/tonic-macros/tests/server.rs similarity index 92% rename from tonic-macros/tests/grpc.rs rename to tonic-macros/tests/server.rs index 401b03a..c26f72d 100644 --- a/tonic-macros/tests/grpc.rs +++ b/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, Status> { println!("Got a request: {:?}", request); diff --git a/tonic/Cargo.toml b/tonic/Cargo.toml index 74d9a0e..0220f86 100644 --- a/tonic/Cargo.toml +++ b/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" } diff --git a/tonic/src/lib.rs b/tonic/src/lib.rs index a738c9b..fe00055 100644 --- a/tonic/src/lib.rs +++ b/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; diff --git a/tower-h2/Cargo.toml b/tower-h2/Cargo.toml new file mode 100644 index 0000000..ee1f254 --- /dev/null +++ b/tower-h2/Cargo.toml @@ -0,0 +1,23 @@ +[package] +name = "tower-h2" +version = "0.1.0" +authors = ["Lucio Franco "] +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" diff --git a/tower-h2/examples/client.rs b/tower-h2/examples/client.rs new file mode 100644 index 0000000..fbba759 --- /dev/null +++ b/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> { + 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); + +impl From> for Body { + fn from(t: Vec) -> Self { + Body(t) + } +} + +impl BufStream for Body { + type Item = std::io::Cursor>; + type Error = std::io::Error; + + fn poll_buf(&mut self, _cx: &mut Context<'_>) -> Poll>> { + 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() + } +} diff --git a/tower-h2/examples/server.rs b/tower-h2/examples/server.rs new file mode 100644 index 0000000..d0859ce --- /dev/null +++ b/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> for Svc { + type Response = Response; + type Error = h2::Error; + type Future = future::Ready>; + + fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll> { + Ok(()).into() + } + + fn call(&mut self, req: Request) -> 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>; + + fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll> { + Ok(()).into() + } + + fn call(&mut self, _: ()) -> Self::Future { + future::ok(Svc) + } +} + +#[tokio::main] +async fn main() -> Result<(), Box> { + 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); + +impl From> for Body { + fn from(t: Vec) -> Self { + Body(t) + } +} + +impl BufStream for Body { + type Item = std::io::Cursor>; + type Error = std::io::Error; + + fn poll_buf(&mut self, _cx: &mut Context<'_>) -> Poll>> { + 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() + } +} diff --git a/tower-h2/src/buf.rs b/tower-h2/src/buf.rs new file mode 100644 index 0000000..80ae201 --- /dev/null +++ b/tower-h2/src/buf.rs @@ -0,0 +1,38 @@ +use bytes::Buf; + +pub struct SendBuf { + inner: Option, +} + +impl SendBuf { + pub fn new(buf: T) -> SendBuf { + SendBuf { inner: Some(buf) } + } + + pub fn none() -> SendBuf { + SendBuf { inner: None } + } +} + +impl Buf for SendBuf { + 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 => {} + } + } +} diff --git a/tower-h2/src/client.rs b/tower-h2/src/client.rs new file mode 100644 index 0000000..0c0f902 --- /dev/null +++ b/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 = Pin + Send + 'static>>; + +pub struct Connection +where + B: Body + Unpin, + B::Data: Unpin, +{ + client: SendRequest>, +} + +impl Connection +where + B: Body + Send + Unpin + 'static, + B::Data: Send + Unpin + 'static, + B::Error: Into>, +{ + pub async fn handshake(io: T) -> Result, 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) -> Result, h2::Error> { + future::poll_fn(|cx| self.poll_ready(cx)).await?; + + self.call(request).await + } +} + +impl Service> for Connection +where + B: Body + Send + Unpin + 'static, + B::Data: Send + Unpin + 'static, + B::Error: Into>, +{ + type Response = Response; + type Error = h2::Error; + type Future = BoxFuture>; + + fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll> { + self.client.poll_ready(cx) + } + + fn call(&mut self, request: Request) -> 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))) + } +} diff --git a/tower-h2/src/error.rs b/tower-h2/src/error.rs new file mode 100644 index 0000000..201340d --- /dev/null +++ b/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::() { + return h2_err.reason().unwrap_or(h2::Reason::INTERNAL_ERROR); + } + cause = err.source(); + } + + // unknown error + h2::Reason::INTERNAL_ERROR +} diff --git a/tower-h2/src/flush.rs b/tower-h2/src/flush.rs new file mode 100644 index 0000000..e7e8e60 --- /dev/null +++ b/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 +where + S: Body, +{ + h2: SendStream>, + body: S, + state: FlushState, +} + +enum FlushState { + Data, + Trailers, + Done, +} + +enum DataOrTrailers { + Data(B), + Trailers(HeaderMap), +} + +// ===== impl Flush ===== + +impl Flush +where + S: Body, + S::Error: Into>, +{ + pub fn new(src: S, dst: SendStream>) -> Self { + Flush { + h2: dst, + body: src, + state: FlushState::Data, + } + } + + /// Try to flush the body. + fn poll_complete(&mut self, cx: &mut Context<'_>) -> Poll> { + 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, 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 Future for Flush +where + S: Body + Unpin, + S::Error: Into>, +{ + type Output = Result<(), ()>; + + fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + Pin::new(&mut self) + .poll_complete(cx) + .map_err(|err| warn!("error flushing stream: {:?}", err)) + } +} diff --git a/tower-h2/src/lib.rs b/tower-h2/src/lib.rs new file mode 100644 index 0000000..4a8cad4 --- /dev/null +++ b/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; diff --git a/tower-h2/src/recv_body.rs b/tower-h2/src/recv_body.rs new file mode 100644 index 0000000..64306a5 --- /dev/null +++ b/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>> { + 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, 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 for Bytes { + fn from(src: Data) -> Self { + src.bytes + } +} + +impl From for BytesMut { + fn from(src: Data) -> Self { + src.bytes.into() + } +} diff --git a/tower-h2/src/server.rs b/tower-h2/src/server.rs new file mode 100644 index 0000000..889b717 --- /dev/null +++ b/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 +where + M: MakeService<(), Request>, + B: Body, +{ + maker: M, + builder: h2::server::Builder, + _pd: PhantomData, +} + +impl Server +where + M: MakeService<(), Request, Response = Response>, + M::MakeError: Into>, + M::Error: Into>, + B: Body + Send + Unpin + 'static, + B::Data: Send + Unpin, + B::Error: Into>, +{ + pub fn new(maker: M, builder: h2::server::Builder) -> Self { + Self { + maker, + builder, + _pd: PhantomData + } + } + + pub async fn serve(&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> = + 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( + response: Response, + mut send_response: h2::server::SendResponse>, + ) where + B: Body + Send + Unpin + 'static, + B::Data: Unpin, + B::Error: Into>, + { + 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); + } + } + }