From f750dfa11143a76827084e9dca28c157061daee4 Mon Sep 17 00:00:00 2001 From: Lucio Franco Date: Wed, 28 Aug 2019 17:24:53 -0400 Subject: [PATCH] Fix interoptests and rewrite decode --- Cargo.toml | 1 - tonic-examples/Cargo.toml | 8 +- tonic-examples/src/helloworld/client.rs | 15 +- tonic-examples/src/helloworld/server.rs | 20 +- tonic-examples/src/routeguide/client.rs | 15 +- tonic-examples/src/routeguide/server.rs | 24 +- tonic-interop/Cargo.toml | 6 +- tonic-interop/src/bin/client.rs | 8 +- tonic-interop/src/bin/server.rs | 27 +- tonic-interop/src/client.rs | 25 +- tonic-macros/Cargo.toml | 4 +- tonic-macros/src/client.rs | 4 + tonic-macros/src/lib.rs | 6 +- tonic-macros/src/service.rs | 4 +- tonic/Cargo.toml | 20 +- tonic/src/body.rs | 72 +++- tonic/src/client/grpc.rs | 52 ++- tonic/src/codec/decode.rs | 329 +++++++++--------- tonic/src/codec/mod.rs | 2 +- tonic/src/lib.rs | 4 +- tonic/src/server/grpc.rs | 25 +- .../src => tonic/src/service}/add_origin.rs | 0 tonic/src/{service.rs => service/mod.rs} | 2 + tower-h2/Cargo.toml | 26 -- tower-h2/examples/client.rs | 57 --- tower-h2/examples/server.rs | 111 ------ tower-h2/src/buf.rs | 38 -- tower-h2/src/client.rs | 81 ----- tower-h2/src/error.rs | 12 - tower-h2/src/flush.rs | 196 ----------- tower-h2/src/lib.rs | 15 - tower-h2/src/recv_body.rs | 98 ------ tower-h2/src/server.rs | 110 ------ 33 files changed, 370 insertions(+), 1047 deletions(-) rename {tower-h2/src => tonic/src/service}/add_origin.rs (100%) rename tonic/src/{service.rs => service/mod.rs} (98%) delete mode 100644 tower-h2/Cargo.toml delete mode 100644 tower-h2/examples/client.rs delete mode 100644 tower-h2/examples/server.rs delete mode 100644 tower-h2/src/buf.rs delete mode 100644 tower-h2/src/client.rs delete mode 100644 tower-h2/src/error.rs delete mode 100644 tower-h2/src/flush.rs delete mode 100644 tower-h2/src/lib.rs delete mode 100644 tower-h2/src/recv_body.rs delete mode 100644 tower-h2/src/server.rs diff --git a/Cargo.toml b/Cargo.toml index 9cc7739..9dfd391 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -5,5 +5,4 @@ members = [ "tonic-build", "tonic-examples", "tonic-interop", - "tower-h2" ] diff --git a/tonic-examples/Cargo.toml b/tonic-examples/Cargo.toml index 778ddce..3ccb7ea 100644 --- a/tonic-examples/Cargo.toml +++ b/tonic-examples/Cargo.toml @@ -24,15 +24,15 @@ path = "src/routeguide/client.rs" [dependencies] tonic = { path = "../tonic" } -tower-h2 = { path = "../tower-h2" } -futures-preview = { version = "=0.3.0-alpha.17", default-features = false, features = ["alloc"]} -tokio = "=0.2.0-alpha.1" +hyper = { path = "../../hyper" } +futures-preview = { version = "=0.3.0-alpha.18", default-features = false, features = ["alloc"]} +tokio = "=0.2.0-alpha.2" prost = "0.5" prost-derive = "0.5" bytes = "0.4" serde_json = "1.0" serde = { version = "1.0", features = ["derive"] } -async-stream = "0.1" +async-stream = { path = "../../async-stream/async-stream" } http = "0.1" [build-dependencies] diff --git a/tonic-examples/src/helloworld/client.rs b/tonic-examples/src/helloworld/client.rs index 9c30b9b..c96119f 100644 --- a/tonic-examples/src/helloworld/client.rs +++ b/tonic-examples/src/helloworld/client.rs @@ -1,5 +1,7 @@ -use tokio::net::TcpStream; -use tower_h2::{add_origin::AddOrigin, Connection}; +use hyper::client::conn::Builder; +use hyper::client::connect::HttpConnector; +use hyper::client::service::{Connect, MakeService}; +use tonic::service::add_origin::AddOrigin; pub mod hello_world { include!(concat!(env!("OUT_DIR"), "/helloworld.rs")); @@ -8,12 +10,13 @@ pub mod hello_world { #[tokio::main] async fn main() -> Result<(), Box> { - let addr = "[::1]:50051".parse()?; - let io = TcpStream::connect(&addr).await?; + let origin = http::Uri::from_static("http://[::1]:50051"); - let origin = http::Uri::from_shared(format!("http://{}", addr).into()).unwrap(); + let settings = Builder::new().http2_only(true).clone(); + let mut maker = Connect::new(HttpConnector::new(1), settings); + + let svc = maker.make_service(origin.clone()).await?; - let svc = Connection::handshake(io).await?; let svc = AddOrigin::new(svc, origin); let mut client = hello_world::GreeterClient::new(svc); diff --git a/tonic-examples/src/helloworld/server.rs b/tonic-examples/src/helloworld/server.rs index d1aa570..80fa85e 100644 --- a/tonic-examples/src/helloworld/server.rs +++ b/tonic-examples/src/helloworld/server.rs @@ -1,7 +1,7 @@ +use hyper::Server; use std::time::Duration; -use tokio::{net::TcpListener, timer::Delay}; +use tokio::timer::Delay; use tonic::{Request, Response, Status}; -use tower_h2::Server; pub mod hello_world { include!(concat!(env!("OUT_DIR"), "/helloworld.rs")); @@ -39,20 +39,12 @@ impl MyGreeter { #[tokio::main] async fn main() -> Result<(), Box> { let addr = "[::1]:50051".parse().unwrap(); - let mut bind = TcpListener::bind(&addr)?; - let greeter = MyGreeter::default(); - let mut server = Server::new(GreeterServer::new(greeter), 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); - } - } + Server::bind(&addr) + .http2_only(true) + .serve(GreeterServer::new(greeter)) + .await?; Ok(()) } diff --git a/tonic-examples/src/routeguide/client.rs b/tonic-examples/src/routeguide/client.rs index 95944d5..173a4d0 100644 --- a/tonic-examples/src/routeguide/client.rs +++ b/tonic-examples/src/routeguide/client.rs @@ -1,9 +1,12 @@ use futures::TryStreamExt; +use hyper::client::conn::Builder; +use hyper::client::connect::HttpConnector; +use hyper::client::service::{Connect, MakeService}; use route_guide::{Point, RouteNote}; use std::time::{Duration, Instant}; -use tokio::{net::TcpStream, timer::Interval}; +use tokio::timer::Interval; +use tonic::service::add_origin::AddOrigin; use tonic::Request; -use tower_h2::{add_origin::AddOrigin, Connection}; mod route_guide { include!(concat!(env!("OUT_DIR"), "/routeguide.rs")); @@ -12,12 +15,12 @@ mod route_guide { #[tokio::main] async fn main() -> Result<(), Box> { - let addr = "[::1]:10000".parse()?; - let io = TcpStream::connect(&addr).await?; + let origin = http::Uri::from_static("http://[::1]:10000"); - let origin = http::Uri::from_shared(format!("http://{}", addr).into()).unwrap(); + let settings = Builder::new().http2_only(true).clone(); + let mut maker = Connect::new(HttpConnector::new(1), settings); - let svc = Connection::handshake(io).await?; + let svc = maker.make_service(origin.clone()).await?; let svc = AddOrigin::new(svc, origin); let mut client = route_guide::RouteGuideClient::new(svc); diff --git a/tonic-examples/src/routeguide/server.rs b/tonic-examples/src/routeguide/server.rs index a89fb3c..645e97f 100644 --- a/tonic-examples/src/routeguide/server.rs +++ b/tonic-examples/src/routeguide/server.rs @@ -1,16 +1,13 @@ mod data; use futures::{Stream, StreamExt}; +use hyper::Server; use std::collections::HashMap; use std::hash::{Hash, Hasher}; use std::sync::Arc; use std::time::Instant; -use tokio::{ - net::TcpListener, - sync::{mpsc, Lock}, -}; +use tokio::sync::{mpsc, Lock}; use tonic::{Request, Response, Status}; -use tower_h2::Server; pub mod routeguide { include!(concat!(env!("OUT_DIR"), "/routeguide.rs")); @@ -151,9 +148,8 @@ impl RouteGuide { #[tokio::main] async fn main() -> Result<(), Box> { let addr = "[::1]:10000".parse().unwrap(); - let mut bind = TcpListener::bind(&addr)?; - println!("Listening on: {}", bind.local_addr()?); + println!("Listening on: {}", addr); let route_guide = RouteGuide { state: State { @@ -162,17 +158,11 @@ async fn main() -> Result<(), Box> { notes: Lock::new(HashMap::new()), }, }; - let mut server = Server::new(RouteGuideServer::new(route_guide), 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); - } - } + Server::bind(&addr) + .http2_only(true) + .serve(RouteGuideServer::new(route_guide)) + .await?; Ok(()) } diff --git a/tonic-interop/Cargo.toml b/tonic-interop/Cargo.toml index 74f884f..f92d6fa 100644 --- a/tonic-interop/Cargo.toml +++ b/tonic-interop/Cargo.toml @@ -13,14 +13,14 @@ name = "server" path = "src/bin/server.rs" [dependencies] -tokio = "=0.2.0-alpha.1" +tokio = "=0.2.0-alpha.2" tonic = { path = "../tonic" } prost = "0.5" prost-derive = "0.5" bytes = "0.4" -tower-h2 = { path = "../tower-h2" } http = "0.1" -futures-util-preview = "=0.3.0-alpha.17" +futures-util-preview = "=0.3.0-alpha.18" +hyper = { path = "../../hyper" } console = "0.7" structopt = "0.2" diff --git a/tonic-interop/src/bin/client.rs b/tonic-interop/src/bin/client.rs index a3404ed..e812923 100644 --- a/tonic-interop/src/bin/client.rs +++ b/tonic-interop/src/bin/client.rs @@ -1,5 +1,5 @@ -use tonic_interop::client; use structopt::{clap::arg_enum, StructOpt}; +use tonic_interop::client; #[derive(StructOpt)] struct Opts { @@ -9,7 +9,7 @@ struct Opts { min_values = 1, raw(possible_values = r#"&Testcase::variants()"#) )] - test_case: Vec + test_case: Vec, } #[tokio::main] @@ -52,7 +52,9 @@ async fn main() -> Result<(), Box> { Testcase::unimplemented_service => { client::unimplemented_service(&mut unimplemented_client, &mut test_results).await } - Testcase::custom_metadata => client::custom_metadata(&mut client, &mut test_results).await, + Testcase::custom_metadata => { + client::custom_metadata(&mut client, &mut test_results).await + } _ => unimplemented!(), } diff --git a/tonic-interop/src/bin/server.rs b/tonic-interop/src/bin/server.rs index dc57f91..c8ea6ad 100644 --- a/tonic-interop/src/bin/server.rs +++ b/tonic-interop/src/bin/server.rs @@ -1,6 +1,5 @@ -use tokio::net::TcpListener; +use hyper::Server; use tonic::{Code, Request, Response, Status}; -use tower_h2::{Server, Builder}; pub mod pb { #![allow(dead_code)] @@ -18,8 +17,8 @@ pub struct TestService { #[tonic::server(service = "grpc.testing.TestService", proto = "pb")] impl TestService { - pub async fn empty_call(&self, request: Request) -> Result, Status> { - println!("empty_call; REQUEST={:?}", request); + pub async fn empty_call(&self, _request: Request) -> Result, Status> { + println!("empty_call"); Ok(Response::new(Empty {})) } @@ -27,7 +26,7 @@ impl TestService { &self, request: Request, ) -> Result, Status> { - println!("unary_call; REQUEST={:?}", request); + println!("unary_call"); let req = request.into_inner(); @@ -59,23 +58,13 @@ impl TestService { async fn main() -> Result<(), Box> { pretty_env_logger::init(); let addr = "127.0.0.1:10000".parse().unwrap(); - let mut bind = TcpListener::bind(&addr)?; let greeter = TestService::default(); - let mut settings = Builder::default(); - settings.initial_connection_window_size(1_000_000_000); - let mut server = Server::new(TestServiceServer::new(greeter), Default::default()); - while let Ok((sock, _addr)) = bind.accept().await { - println!("new connection"); - if let Err(e) = sock.set_nodelay(true) { - return Err(e.into()); - } - - if let Err(e) = server.serve(sock).await { - println!("H2 ERROR: {}", e); - } - } + Server::bind(&addr) + .http2_only(true) + .serve(TestServiceServer::new(greeter)) + .await?; Ok(()) } diff --git a/tonic-interop/src/client.rs b/tonic-interop/src/client.rs index 7e589ae..7d46f57 100644 --- a/tonic-interop/src/client.rs +++ b/tonic-interop/src/client.rs @@ -1,12 +1,15 @@ use crate::{pb::*, test_assert, TestAssertion}; use futures_util::{future, stream, SinkExt, StreamExt}; +use hyper::client::conn::{Builder, SendRequest}; +use hyper::client::connect::HttpConnector; +use hyper::client::service::{Connect, MakeService}; use std::net::SocketAddr; -use tokio::{net::TcpStream, sync::mpsc}; +use tokio::sync::mpsc; +use tonic::service::add_origin::AddOrigin; use tonic::{metadata::MetadataValue, Code, Request, Response, Status}; -use tower_h2::{add_origin::AddOrigin, Connection}; -pub type Client = TestServiceClient>>; -pub type UnimplementedClient = UnimplementedServiceClient>>; +pub type Client = TestServiceClient>>; +pub type UnimplementedClient = UnimplementedServiceClient>>; tonic::client!(service = "grpc.testing.TestService", proto = "crate::pb"); tonic::client!( @@ -23,11 +26,12 @@ const SPECIAL_TEST_STATUS_MESSAGE: &'static str = "\t\ntest with whitespace\r\nand Unicode BMP ☺ and non-BMP 😈\t\n"; pub async fn create(addr: SocketAddr) -> Result> { - let io = TcpStream::connect(&addr).await?; - let origin = http::Uri::from_shared(format!("http://{}", addr).into()).unwrap(); - let svc = Connection::handshake(io).await?; + let settings = Builder::new().http2_only(true).clone(); + let mut maker = Connect::new(HttpConnector::new(1), settings); + + let svc = maker.make_service(origin.clone()).await?; let svc = AddOrigin::new(svc, origin); Ok(TestServiceClient::new(svc)) @@ -36,11 +40,12 @@ pub async fn create(addr: SocketAddr) -> Result Result> { - let io = TcpStream::connect(&addr).await?; - let origin = http::Uri::from_shared(format!("http://{}", addr).into()).unwrap(); - let svc = Connection::handshake(io).await?; + let settings = Builder::new().http2_only(true).clone(); + let mut maker = Connect::new(HttpConnector::new(1), settings); + + let svc = maker.make_service(origin.clone()).await?; let svc = AddOrigin::new(svc, origin); Ok(UnimplementedServiceClient::new(svc)) diff --git a/tonic-macros/Cargo.toml b/tonic-macros/Cargo.toml index 1d12a2a..4be076b 100644 --- a/tonic-macros/Cargo.toml +++ b/tonic-macros/Cargo.toml @@ -15,6 +15,6 @@ serde_json = "1.0" serde = { version = "1.0", features = ["derive"] } [dev-dependencies] -tokio = "=0.2.0-alpha.1" +tokio = "=0.2.0-alpha.2" tonic = { path = "../tonic" } -futures-preview = "=0.3.0-alpha.17" +futures-preview = "=0.3.0-alpha.18" diff --git a/tonic-macros/src/client.rs b/tonic-macros/src/client.rs index 5d302f6..7eea05a 100644 --- a/tonic-macros/src/client.rs +++ b/tonic-macros/src/client.rs @@ -33,6 +33,7 @@ fn generate_unary(method: &Method, proto: &str, path: String) -> TokenStream { quote! { pub async fn #ident (&mut self, request: tonic::Request<#request>) -> Result, tonic::Status> { + self.inner.ready().await?; let codec = tonic::codec::ProstCodec::new(); let path = http::uri::PathAndQuery::from_static(#path); self.inner.unary(request, path, codec).await @@ -48,6 +49,7 @@ fn generate_server_streaming(method: &Method, proto: &str, path: String) -> Toke quote! { pub async fn #ident (&mut self, request: tonic::Request<#request>) -> Result>, tonic::Status> { + self.inner.ready().await?; let codec = tonic::codec::ProstCodec::new(); let path = http::uri::PathAndQuery::from_static(#path); self.inner.server_streaming(request, path, codec).await @@ -65,6 +67,7 @@ fn generate_client_streaming(method: &Method, proto: &str, path: String) -> Toke -> Result, tonic::Status> where S: tonic::_codegen::Stream> + Send + 'static, { + self.inner.ready().await?; let codec = tonic::codec::ProstCodec::new(); let path = http::uri::PathAndQuery::from_static(#path); let request = request.map(|s| Box::pin(s)); @@ -83,6 +86,7 @@ fn generate_streaming(method: &Method, proto: &str, path: String) -> TokenStream -> Result>, tonic::Status> where S: tonic::_codegen::Stream> + Send + 'static, { + self.inner.ready().await?; let codec = tonic::codec::ProstCodec::new(); let path = http::uri::PathAndQuery::from_static(#path); let request = request.map(|s| Box::pin(s)); diff --git a/tonic-macros/src/lib.rs b/tonic-macros/src/lib.rs index e5147b4..eca2d0f 100644 --- a/tonic-macros/src/lib.rs +++ b/tonic-macros/src/lib.rs @@ -29,12 +29,16 @@ pub fn client(attr: TokenStream) -> TokenStream { where T: tonic::GrpcService, T::ResponseBody: tonic::body::Body + tonic::_codegen::HttpBody + Send + 'static, ::Error: Into + Send, - ::Data: Send, { + ::Data: Into + Send, { pub fn new(inner: T) -> Self { let inner = tonic::client::Grpc::new(inner); Self { inner } } + pub async fn ready(&mut self) -> Result<(), tonic::Status> { + self.inner.ready().await + } + #methods } diff --git a/tonic-macros/src/service.rs b/tonic-macros/src/service.rs index 29440f7..553cfe4 100644 --- a/tonic-macros/src/service.rs +++ b/tonic-macros/src/service.rs @@ -106,7 +106,7 @@ pub(crate) fn generate(service: ServiceDef) -> TokenStream { } } - impl Service> for #server_service { + impl Service> for #server_service { type Response = http::Response; type Error = tonic::error::Never; type Future = BoxFuture; @@ -115,7 +115,7 @@ pub(crate) fn generate(service: ServiceDef) -> TokenStream { Poll::Ready(Ok(())) } - fn call(&mut self, req: http::Request) -> Self::Future { + fn call(&mut self, req: http::Request) -> Self::Future { let inner = self.inner.clone(); match req.uri().path() { diff --git a/tonic/Cargo.toml b/tonic/Cargo.toml index a0827f2..943c8f3 100644 --- a/tonic/Cargo.toml +++ b/tonic/Cargo.toml @@ -7,8 +7,8 @@ edition = "2018" # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html [dependencies] -futures-core-preview = "=0.3.0-alpha.17" -futures-util-preview = "=0.3.0-alpha.17" +futures-core-preview = "=0.3.0-alpha.18" +futures-util-preview = "=0.3.0-alpha.18" tonic-macros = { path = "../tonic-macros" } tracing = "0.1" http = "0.1.14" @@ -16,15 +16,9 @@ base64 = "0.10" bytes = "0.4.7" prost = "0.5" percent-encoding = "1.0.1" -tower-service = { git = "https://github.com/tower-rs/tower", branch = "std-future" } -tokio-codec = "=0.2.0-alpha.1" -async-stream = "0.1.0" -# http-body = { git = "https://github.com/hyperium/http-body" } -http-body = { path = "../../http-body" } +tower-service = "=0.3.0-alpha.1" +tokio-codec = "=0.2.0-alpha.2" +# async-stream = "0.1.0" +async-stream = { path = "../../async-stream/async-stream" } +http-body = "0.2.0-alpha.1" pin-project = "0.4.0-alpha.2" - -[dev-dependencies] -tokio = "=0.2.0-alpha.1" -tokio-buf = "=0.2.0-alpha.1" -prost-derive = "0.5" -tower-h2 = { path = "../tower-h2" } diff --git a/tonic/src/body.rs b/tonic/src/body.rs index c4f4ef7..04cf028 100644 --- a/tonic/src/body.rs +++ b/tonic/src/body.rs @@ -62,19 +62,33 @@ mod sealed { } pub struct BoxBody { - inner: Pin + Send + 'static>>, + inner: Pin + Send + 'static>>, } +struct MapBody(B); + impl BoxBody { /// Create a new `BoxBody` mapping item and error to the default types. - pub fn map_from(inner: B) -> Self + pub fn new(inner: B) -> Self where - B: HttpBody + Send + 'static, + B: Body + Send + 'static, { BoxBody { inner: Box::pin(inner), } } + + /// Create a new `BoxBody` mapping item and error to the default types. + pub fn map_from(inner: B) -> Self + where + B: Body + Send + 'static, + B::Data: Into, + B::Error: Into, + { + BoxBody { + inner: Box::pin(MapBody(inner)), + } + } } impl HttpBody for BoxBody { @@ -82,20 +96,66 @@ impl HttpBody for BoxBody { type Error = Status; fn is_end_stream(&self) -> bool { - HttpBody::is_end_stream(&self.inner) + // Body::is_end_stream(&self.inner) + self.inner.is_end_stream() } fn poll_data( mut self: Pin<&mut Self>, cx: &mut Context<'_>, ) -> Poll>> { - HttpBody::poll_data(self.inner.as_mut(), cx) + Body::poll_data(self.inner.as_mut(), cx) } fn poll_trailers( mut self: Pin<&mut Self>, cx: &mut Context<'_>, ) -> Poll, Self::Error>> { - HttpBody::poll_trailers(self.inner.as_mut(), cx) + Body::poll_trailers(self.inner.as_mut(), cx) + } +} + +impl HttpBody for MapBody +where + B: Body, + B::Data: Into, + B::Error: Into, +{ + type Data = BytesBuf; + type Error = Status; + + fn is_end_stream(&self) -> bool { + self.0.is_end_stream() + } + + fn poll_data( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + ) -> Poll>> { + let v = unsafe { + let me = self.get_unchecked_mut(); + Pin::new_unchecked(&mut me.0).poll_data(cx) + }; + match futures_util::ready!(v) { + Some(Ok(i)) => Poll::Ready(Some(Ok(i.into().into_buf()))), + Some(Err(e)) => { + let err = Status::map_error(e.into()); + Poll::Ready(Some(Err(err))) + } + None => Poll::Ready(None), + } + } + + fn poll_trailers( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + ) -> Poll, Self::Error>> { + let v = unsafe { + let me = self.get_unchecked_mut(); + Pin::new_unchecked(&mut me.0).poll_trailers(cx) + }; + + let v = futures_util::ready!(v).map_err(|e| Status::from_error(&*e.into())); + Poll::Ready(v) } } diff --git a/tonic/src/client/grpc.rs b/tonic/src/client/grpc.rs index 02c94ba..0957802 100644 --- a/tonic/src/client/grpc.rs +++ b/tonic/src/client/grpc.rs @@ -1,8 +1,9 @@ use crate::{ body::{Body, BoxBody}, - codec::{decode_empty, decode_response, encode_client, Codec, Streaming}, + codec::{encode_client, Codec, Streaming}, Code, GrpcService, Request, Response, Status, }; +use bytes::Bytes; use futures_core::Stream; use futures_util::{future, stream, TryStreamExt}; use http::{ @@ -20,6 +21,23 @@ impl Grpc { Self { inner } } + pub async fn ready(&mut self) -> Result<(), Status> + where + T: GrpcService, + T::ResponseBody: Body + HttpBody + Send + 'static, + ::Error: Into + Send, + ::Data: Send, + { + futures_util::future::poll_fn(|cx| self.inner.poll_ready(cx)) + .await + .map_err(|e| { + Status::new( + Code::Unknown, + format!("Unexpected connection error: {}", e.into()), + ) + }) + } + pub async fn unary( &mut self, request: Request, @@ -30,7 +48,7 @@ impl Grpc { T: GrpcService, T::ResponseBody: Body + HttpBody + Send + 'static, ::Error: Into + Send, - ::Data: Send, + ::Data: Into + Send, C: Codec, C::Encoder: Send + 'static, C::Decoder: Send + 'static, @@ -51,7 +69,7 @@ impl Grpc { T: GrpcService, T::ResponseBody: Body + HttpBody + Send + 'static, ::Error: Into + Send, - ::Data: Send, + ::Data: Into + Send, S: Stream> + Send + 'static, C: Codec, C::Encoder: Send + 'static, @@ -59,7 +77,9 @@ impl Grpc { M1: Send, M2: Send + Unpin + 'static, { - let (parts, mut body) = self.streaming(request, path, codec).await?.into_parts(); + let (parts, body) = self.streaming(request, path, codec).await?.into_parts(); + + futures_util::pin_mut!(body); let message = body .try_next() @@ -79,7 +99,7 @@ impl Grpc { T: GrpcService, T::ResponseBody: Body + HttpBody + Send + 'static, ::Error: Into + Send, - ::Data: Send, + ::Data: Into + Send, C: Codec, C::Encoder: Send + 'static, C::Decoder: Send + 'static, @@ -100,7 +120,7 @@ impl Grpc { T: GrpcService, T::ResponseBody: Body + HttpBody + Send + 'static, ::Error: Into + Send, - ::Data: Send, + ::Data: Into + Send, S: Stream> + Send + 'static, C: Codec, C::Encoder: Send + 'static, @@ -115,7 +135,7 @@ impl Grpc { let request = request .map(|s| encode_client(codec.encoder(), Box::pin(s))) - .map(BoxBody::map_from); + .map(BoxBody::new); let mut request = request.into_http(uri); @@ -151,17 +171,13 @@ impl Grpc { true }; - let response = response - .map(|b| { - if expect_additional_trailers { - future::Either::Left( - decode_response(codec.decoder(), b, status_code).into_stream(), - ) - } else { - future::Either::Right(decode_empty(codec.decoder(), b).into_stream()) - } - }) - .map(Streaming::new); + let response = response.map(|body| { + if expect_additional_trailers { + Streaming::new_response(codec.decoder(), body, status_code) + } else { + Streaming::new_empty(codec.decoder(), body) + } + }); Ok(Response::from_http(response)) } diff --git a/tonic/src/codec/decode.rs b/tonic/src/codec/decode.rs index 9ed8e02..998bd5e 100644 --- a/tonic/src/codec/decode.rs +++ b/tonic/src/codec/decode.rs @@ -1,79 +1,27 @@ +use super::Decoder; +use crate::body::BoxBody; +use crate::metadata::MetadataMap; use crate::{Code, Status}; -use bytes::{Buf, BufMut, BytesMut, IntoBuf}; -use futures_core::{Stream, TryStream}; -use futures_util::future; +use bytes::{Buf, BufMut, Bytes, BytesMut, IntoBuf}; +use futures_core::Stream; +use futures_util::{future, ready}; use http::StatusCode; use http_body::Body; use std::fmt; use std::pin::Pin; use std::task::{Context, Poll}; -use tokio_codec::Decoder; use tracing::{debug, trace}; -pub fn decode_request( - decoder: T, - source: B, -) -> impl TryStream + 'static -where - T: Decoder + 'static, - T::Item: Unpin + 'static, - B: Body + 'static, - B::Error: Into, -{ - decode(decoder, source, Direction::Request) -} - -pub fn decode_response( - decoder: T, - source: B, - status: StatusCode, -) -> impl TryStream + 'static -where - T: Decoder + 'static, - T::Item: Unpin + 'static, - B: Body + 'static, - B::Error: Into, -{ - decode(decoder, source, Direction::Response(status)) -} - -pub fn decode_empty( - decoder: T, - source: B, -) -> impl TryStream + 'static -where - T: Decoder + 'static, - T::Item: Unpin + 'static, - B: Body + 'static, - B::Error: Into, -{ - decode(decoder, source, Direction::EmptyResponse) -} - +// #[derive(Debug)] pub struct Streaming { - inner: Pin> + Send + 'static>>, + decoder: Box + Send + 'static>, + body: BoxBody, + state: State, + direction: Direction, + buf: BytesMut, } -impl Streaming { - pub fn new(inner: impl Stream> + Send + 'static) -> Self { - let inner = Box::pin(inner); - Self { inner } - } -} - -impl Stream for Streaming { - type Item = Result; - - fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - Pin::new(&mut self.inner).poll_next(cx) - } -} - -impl fmt::Debug for Streaming { - fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { - write!(f, "Streaming") - } -} +impl Unpin for Streaming {} #[derive(Debug)] enum State { @@ -88,50 +36,160 @@ enum Direction { EmptyResponse, } -fn decode( - mut decoder: T, - mut source: B, - direction: Direction, -) -> impl TryStream + 'static -where - T: Decoder + 'static, - T::Item: Unpin + 'static, - B: Body + 'static, - B::Error: Into, -{ - async_stream::try_stream! { - let mut buf = BytesMut::with_capacity(1024 * 1024 * 1024); - let mut state = State::ReadHeader; +impl Streaming { + pub fn new_response(decoder: D, body: B, status_code: StatusCode) -> Self + where + B: Body + Send + 'static, + B::Data: Into, + B::Error: Into, + D: Decoder + Send + 'static, + { + Self { + decoder: Box::new(decoder), + body: BoxBody::map_from(body), + state: State::ReadHeader, + direction: Direction::Response(status_code), + // FIXME: update this with a reasonable size + buf: BytesMut::with_capacity(1024 * 1024), + } + } + pub fn new_empty(decoder: D, body: B) -> Self + where + B: Body + Send + 'static, + B::Data: Into, + B::Error: Into, + D: Decoder + Send + 'static, + { + Self { + decoder: Box::new(decoder), + body: BoxBody::map_from(body), + state: State::ReadHeader, + direction: Direction::EmptyResponse, + // FIXME: update this with a reasonable size + buf: BytesMut::with_capacity(1024 * 1024), + } + } + + pub fn new_request(decoder: D, body: B) -> Self + where + B: Body + Send + 'static, + B::Data: Into, + B::Error: Into, + D: Decoder + Send + 'static, + { + Self { + decoder: Box::new(decoder), + body: BoxBody::map_from(body), + state: State::ReadHeader, + direction: Direction::Request, + // FIXME: update this with a reasonable size + buf: BytesMut::with_capacity(1024 * 1024), + } + } +} + +impl Streaming { + // pub async fn message(&mut self) -> Option> { + // future::poll_fn(|cx| Pin::new(&mut *self).poll_next(cx)).await + // } + + pub async fn trailers(&mut self) -> Result, Status> { + let map = + future::poll_fn(|cx| unsafe { Pin::new_unchecked(&mut self.body) }.poll_trailers(cx)) + .await + .map_err(|e| Status::from_error(&e))?; + Ok(map.map(MetadataMap::from_headers)) + } + + fn decode_chunk(&mut self) -> Result, Status> { + let mut buf = (&self.buf[..]).into_buf(); + + if let State::ReadHeader = self.state { + if buf.remaining() < 5 { + return Ok(None); + } + + let is_compressed = match buf.get_u8() { + 0 => false, + 1 => { + trace!("message compressed, compression not supported yet"); + return Err(Status::new( + Code::Unimplemented, + "Message compressed, compression not supported yet.".to_string(), + )); + } + f => { + trace!("unexpected compression flag"); + return Err(Status::new( + Code::Internal, + format!("Unexpected compression flag: {}", f), + )); + } + }; + let len = buf.get_u32_be() as usize; + + self.state = State::ReadBody { + compression: is_compressed, + len, + } + } + + if let State::ReadBody { len, .. } = &self.state { + if buf.remaining() < *len { + return Ok(None); + } + + // advance past the header + self.buf.advance(5); + + match self.decoder.decode(&mut self.buf) { + Ok(Some(msg)) => { + self.state = State::ReadHeader; + return Ok(Some(msg)); + } + Ok(None) => return Ok(None), + Err(e) => { + return Err(e); + } + } + } + + Ok(None) + } +} + +impl Stream for Streaming { + type Item = Result; + + fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { loop { - if let Some(item) = decode_chunk(&mut decoder, &mut buf, &mut state)? { - // TODO: implement the ability to poll trailers when we _know_ that - // the comnsumer of this stream will only poll for the first message. - // This means we skip the poll_trailers step. - - yield item; + // TODO: implement the ability to poll trailers when we _know_ that + // the comnsumer of this stream will only poll for the first message. + // This means we skip the poll_trailers step. + match self.decode_chunk()? { + Some(item) => return Poll::Ready(Some(Ok(item))), + None => (), } // FIXME: Figure out how to verify that this is safe - let chunk = match future::poll_fn(|cx| unsafe { std::pin::Pin::new_unchecked(&mut source) }.poll_data(cx)).await { - Some(Ok(d)) => { - Some(d) - }, + let chunk = match ready!(unsafe { Pin::new_unchecked(&mut self.body) }.poll_data(cx)) { + Some(Ok(d)) => Some(d), Some(Err(e)) => { - let err = e.into(); + let err: crate::Error = e.into(); debug!("decoder inner stream error: {:?}", err); let status = Status::from_error(&*err); Err(status)?; break; - }, + } None => None, }; if let Some(data) = chunk { - buf.put(data); + self.buf.put(data); } else { // FIXME: get BytesMut to impl `Buf` directlty? - let buf1 = (&buf[..]).into_buf(); + let buf1 = (&self.buf[..]).into_buf(); if buf1.has_remaining() { trace!("unexpected EOF decoding stream"); Err(Status::new( @@ -144,81 +202,28 @@ where } } - if let Direction::Response(status) = direction { - let trailer = future::poll_fn(|cx| unsafe { std::pin::Pin::new_unchecked(&mut source) }.poll_trailers(cx)); - let trailer = match trailer.await { - Ok(trailer) => crate::status::infer_grpc_status(trailer, status)?, + if let Direction::Response(status) = self.direction { + match ready!(unsafe { Pin::new_unchecked(&mut self.body) }.poll_trailers(cx)) { + Ok(trailer) => { + if let Err(e) = crate::status::infer_grpc_status(trailer, status) { + return Some(Err(e)).into(); + } + } Err(e) => { - let err = e.into(); + let err: crate::Error = e.into(); debug!("decoder inner trailers error: {:?}", err); let status = Status::from_error(&*err); - Err(status)?; - }, - Ok(None) => return, - }; + return Some(Err(status)).into(); + } + } } + + Poll::Ready(None) } } -fn decode_chunk( - decoder: &mut T, - buf1: &mut BytesMut, - state: &mut State, -) -> Result, Status> -where - T: Decoder, -{ - let mut buf = (&buf1[..]).into_buf(); - - if let State::ReadHeader = state { - if buf.remaining() < 5 { - return Ok(None); - } - - let is_compressed = match buf.get_u8() { - 0 => false, - 1 => { - trace!("message compressed, compression not supported yet"); - return Err(crate::Status::new( - crate::Code::Unimplemented, - "Message compressed, compression not supported yet.".to_string(), - )); - } - f => { - trace!("unexpected compression flag"); - return Err(crate::Status::new( - crate::Code::Internal, - format!("Unexpected compression flag: {}", f), - )); - } - }; - let len = buf.get_u32_be() as usize; - - *state = State::ReadBody { - compression: is_compressed, - len, - } +impl fmt::Debug for Streaming { + fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { + write!(f, "Streaming") } - - if let State::ReadBody { len, .. } = state { - if buf.remaining() < *len { - return Ok(None); - } - - // advance past the header - buf1.advance(5); - - match decoder.decode(buf1) { - Ok(Some(msg)) => { - *state = State::ReadHeader; - return Ok(Some(msg)); - } - Ok(None) => return Ok(None), - Err(e) => { - return Err(e); - } - } - } - - Ok(None) } diff --git a/tonic/src/codec/mod.rs b/tonic/src/codec/mod.rs index 0802dd7..dbf5cd0 100644 --- a/tonic/src/codec/mod.rs +++ b/tonic/src/codec/mod.rs @@ -2,7 +2,7 @@ mod decode; mod encode; mod prost; -pub use self::decode::{decode_empty, decode_request, decode_response, Streaming}; +pub use self::decode::Streaming; pub use self::encode::{encode_client, encode_server, EncodeBody}; pub use self::prost::ProstCodec; diff --git a/tonic/src/lib.rs b/tonic/src/lib.rs index 8a4abce..3dcba65 100644 --- a/tonic/src/lib.rs +++ b/tonic/src/lib.rs @@ -9,10 +9,10 @@ pub mod codec; pub mod error; pub mod metadata; pub mod server; +pub mod service; mod request; mod response; -mod service; mod status; pub use body::BoxBody; @@ -38,7 +38,7 @@ pub trait GrpcInnerService { pub mod _codegen { pub use futures_core::Stream; - pub use futures_util::future::{ok, Ready}; + pub use futures_util::future::{ok, poll_fn, Ready}; pub use http_body::Body as HttpBody; pub use std::future::Future; pub use std::pin::Pin; diff --git a/tonic/src/server/grpc.rs b/tonic/src/server/grpc.rs index 4395706..95eb7ad 100644 --- a/tonic/src/server/grpc.rs +++ b/tonic/src/server/grpc.rs @@ -1,9 +1,10 @@ use crate::{ body::BoxBody, - codec::{decode_request, encode_server, Codec, Streaming}, + codec::{encode_server, Codec, Streaming}, server::{ClientStreamingService, ServerStreamingService, StreamingService, UnaryService}, Code, Request, Response, Status, }; +use bytes::Bytes; use futures_core::TryStream; use futures_util::{future, stream, TryStreamExt}; use http_body::Body; @@ -32,7 +33,7 @@ where where S: UnaryService, B: Body + Send + 'static, - B::Data: Send, + B::Data: Into + Send, B::Error: Into + Send, { let request = match self.map_request_unary(req).await { @@ -62,7 +63,7 @@ where S: ServerStreamingService, S::ResponseStream: Send + 'static, B: Body + Send + 'static, - B::Data: Send, + B::Data: Into + Send, B::Error: Into + Send, { let request = match self.map_request_unary(req).await { @@ -88,7 +89,7 @@ where T::Decode: Send + 'static, T::Decoder: Send + 'static, B: Body + Send + 'static, - B::Data: Send + 'static, + B::Data: Into + Send + 'static, B::Error: Into + Send + 'static, { let request = self.map_request_streaming(req); @@ -108,7 +109,7 @@ where S: StreamingService, Response = T::Encode> + Send, S::ResponseStream: Send + 'static, B: Body + Send + 'static, - B::Data: Send, + B::Data: Into + Send, B::Error: Into + Send, { let request = self.map_request_streaming(req); @@ -122,11 +123,11 @@ where ) -> Result, Status> where B: Body + Send + 'static, - B::Data: Send, + B::Data: Into + Send, B::Error: Into + Send, { let (parts, body) = request.into_parts(); - let stream = decode_request(self.codec.decoder(), body).into_stream(); + let stream = Streaming::new_request(self.codec.decoder(), body); futures_util::pin_mut!(stream); @@ -144,12 +145,10 @@ where ) -> Request> where B: Body + Send + 'static, - B::Data: Send, + B::Data: Into + Send, B::Error: Into + Send, { - Request::from_http( - request.map(|b| Streaming::new(decode_request(self.codec.decoder(), b).into_stream())), - ) + Request::from_http(request.map(|body| Streaming::new_request(self.codec.decoder(), body))) } fn map_response( @@ -173,7 +172,7 @@ where // FIXME: try to return impl Trait? // let body = Box::pin(body) as BoxStream; - http::Response::from_parts(parts, BoxBody::map_from(body)) + http::Response::from_parts(parts, BoxBody::new(body)) } Err(status) => { let status = stream::once(future::err(status)); @@ -185,7 +184,7 @@ where http::header::HeaderValue::from_static(T::CONTENT_TYPE), ); - http::Response::from_parts(parts, BoxBody::map_from(body)) + http::Response::from_parts(parts, BoxBody::new(body)) } } } diff --git a/tower-h2/src/add_origin.rs b/tonic/src/service/add_origin.rs similarity index 100% rename from tower-h2/src/add_origin.rs rename to tonic/src/service/add_origin.rs diff --git a/tonic/src/service.rs b/tonic/src/service/mod.rs similarity index 98% rename from tonic/src/service.rs rename to tonic/src/service/mod.rs index aa28015..ade1bbc 100644 --- a/tonic/src/service.rs +++ b/tonic/src/service/mod.rs @@ -1,3 +1,5 @@ +pub mod add_origin; + use crate::body::Body; use http::{Request, Response}; use http_body::Body as HttpBody; diff --git a/tower-h2/Cargo.toml b/tower-h2/Cargo.toml deleted file mode 100644 index a496803..0000000 --- a/tower-h2/Cargo.toml +++ /dev/null @@ -1,26 +0,0 @@ -[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" } -# h2 = { git = "https://github.com/hyperium/h2" } -h2 = { path = "../../h2" } -http = "0.1" -# http-body = { git = "https://github.com/hyperium/http-body" } -http-body = { path = "../../http-body" } -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 deleted file mode 100644 index 9a5264c..0000000 --- a/tower-h2/examples/client.rs +++ /dev/null @@ -1,57 +0,0 @@ -use http::Request; -use std::pin::Pin; -use std::task::{Context, Poll}; -use tokio::net::TcpStream; -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 http_body::Body for Body { - type Data = std::io::Cursor>; - type Error = std::io::Error; - - fn poll_data( - mut self: Pin<&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() - } - - fn poll_trailers( - self: Pin<&mut Self>, - _cx: &mut Context<'_>, - ) -> Poll, Self::Error>> { - Ok(None).into() - } -} diff --git a/tower-h2/examples/server.rs b/tower-h2/examples/server.rs deleted file mode 100644 index 9041020..0000000 --- a/tower-h2/examples/server.rs +++ /dev/null @@ -1,111 +0,0 @@ -use futures_util::future; -use http::{Request, Response}; -use std::pin::Pin; -use std::task::{Context, Poll}; -use tokio::net::TcpListener; -use tower_h2::{RecvBody, Server}; -use tower_service::Service; - -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 http_body::Body for Body { - type Data = std::io::Cursor>; - type Error = std::io::Error; - - fn poll_data( - mut self: Pin<&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() - } - - fn poll_trailers( - self: Pin<&mut Self>, - _cx: &mut Context<'_>, - ) -> Poll, Self::Error>> { - Ok(None).into() - } -} diff --git a/tower-h2/src/buf.rs b/tower-h2/src/buf.rs deleted file mode 100644 index 80ae201..0000000 --- a/tower-h2/src/buf.rs +++ /dev/null @@ -1,38 +0,0 @@ -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 deleted file mode 100644 index ed517de..0000000 --- a/tower-h2/src/client.rs +++ /dev/null @@ -1,81 +0,0 @@ -use crate::{buf::SendBuf, flush::Flush, recv_body::RecvBody}; -use futures_util::{future, FutureExt, TryFutureExt}; -use h2::client::SendRequest; -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, mut body) = request.into_parts(); - let request = Request::from_parts(parts, ()); - - let eos = Pin::new(&mut 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 deleted file mode 100644 index 201340d..0000000 --- a/tower-h2/src/error.rs +++ /dev/null @@ -1,12 +0,0 @@ -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 deleted file mode 100644 index 87c4d39..0000000 --- a/tower-h2/src/flush.rs +++ /dev/null @@ -1,196 +0,0 @@ -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: Pin + Send + 'static>>, - state: FlushState, -} - -#[derive(Debug)] -enum FlushState { - Data, - Trailers, - Done, -} - -enum DataOrTrailers { - Data(B), - Trailers(HeaderMap), -} - -// ===== impl Flush ===== - -impl Flush -where - S: Body + Send + 'static, - S::Error: Into>, -{ - pub fn new(src: S, dst: SendStream>) -> Self { - Flush { - h2: dst, - body: Box::pin(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 = Pin::new(&mut 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 { - // carllerche/h2#270 is fixed. - loop { - match ready!(self.h2.poll_capacity(cx)) { - Some(Ok(0)) => {} - Some(Ok(_)) => break, - Some(Err(e)) => 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!(Pin::new(&mut 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!(Pin::new(&mut 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 + Send + 'static, - 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 deleted file mode 100644 index d7b0950..0000000 --- a/tower-h2/src/lib.rs +++ /dev/null @@ -1,15 +0,0 @@ -#[macro_use] -extern crate log; - -pub mod add_origin; - -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::{Builder, Server}; diff --git a/tower-h2/src/recv_body.rs b/tower-h2/src/recv_body.rs deleted file mode 100644 index 1f6067b..0000000 --- a/tower-h2/src/recv_body.rs +++ /dev/null @@ -1,98 +0,0 @@ -use bytes::{Buf, Bytes, BytesMut}; -use futures_util::TryStreamExt; -use http_body::Body; -use std::pin::Pin; -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: Pin<&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: Pin<&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 deleted file mode 100644 index 66c5699..0000000 --- a/tower-h2/src/server.rs +++ /dev/null @@ -1,110 +0,0 @@ -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 use h2::server::Builder; - -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, mut body) = response.into_parts(); - - // Check if the response is imemdiately an end-of-stream. - let eos = std::pin::Pin::new(&mut 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); - } - } -}