talksik/tonic
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix interoptests and rewrite decode

Browse Source
This commit is contained in:
Lucio Franco
2019-08-28 17:24:53 -04:00
parent a9e2f0e250
commit f750dfa111
33 changed files with 370 additions and 1047 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.toml
-1
View File
@@ -5,5 +5,4 @@ members = [
"tonic-build",
"tonic-examples",
"tonic-interop",
"tower-h2"
]
tonic-examples/Cargo.toml
+4 -4
View File
@@ -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]
tonic-examples/src/helloworld/client.rs
+9 -6
View File
@@ -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<dyn std::error::Error>> {
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);
tonic-examples/src/helloworld/server.rs
+6 -14
View File
@@ -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<dyn std::error::Error>> {
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(())
}
tonic-examples/src/routeguide/client.rs
+9 -6
View File
@@ -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<dyn std::error::Error>> {
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);
tonic-examples/src/routeguide/server.rs
+7 -17
View File
@@ -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<dyn std::error::Error>> {
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<dyn std::error::Error>> {
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(())
}
tonic-interop/Cargo.toml
+3 -3
View File
@@ -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"
tonic-interop/src/bin/client.rs
+5 -3
View File
@@ -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<Testcase>
test_case: Vec<Testcase>,
}
#[tokio::main]
@@ -52,7 +52,9 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
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!(),
}
tonic-interop/src/bin/server.rs
+8 -19
View File
@@ -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<Empty>) -> Result<Response<Empty>, Status> {
println!("empty_call; REQUEST={:?}", request);
pub async fn empty_call(&self, _request: Request<Empty>) -> Result<Response<Empty>, Status> {
println!("empty_call");
Ok(Response::new(Empty {}))
}
@@ -27,7 +26,7 @@ impl TestService {
&self,
request: Request<SimpleRequest>,
) -> Result<Response<SimpleResponse>, 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<dyn std::error::Error>> {
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(())
}
tonic-interop/src/client.rs
+15 -10
View File
@@ -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<AddOrigin<Connection<tonic::BoxBody>>>;
pub type UnimplementedClient = UnimplementedServiceClient<AddOrigin<Connection<tonic::BoxBody>>>;
pub type Client = TestServiceClient<AddOrigin<SendRequest<tonic::BoxBody>>>;
pub type UnimplementedClient = UnimplementedServiceClient<AddOrigin<SendRequest<tonic::BoxBody>>>;
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<Client, Box<dyn std::error::Error>> {
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<Client, Box<dyn std::error::Erro
pub async fn create_unimplemented(
addr: SocketAddr,
) -> Result<UnimplementedClient, Box<dyn std::error::Error>> {
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))
tonic-macros/Cargo.toml
+2 -2
View File
@@ -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"
tonic-macros/src/client.rs
+4
View File
@@ -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::Response<#response>, 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::Response<tonic::codec::Streaming<#response>>, 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::Response<#response>, tonic::Status>
where S: tonic::_codegen::Stream<Item = Result<#request, tonic::Status>> + 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::Response<tonic::codec::Streaming<#response>>, tonic::Status>
where S: tonic::_codegen::Stream<Item = Result<#request, tonic::Status>> + 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));
tonic-macros/src/lib.rs
+5 -1
View File
@@ -29,12 +29,16 @@ pub fn client(attr: TokenStream) -> TokenStream {
where T: tonic::GrpcService<tonic::body::BoxBody>,
T::ResponseBody: tonic::body::Body + tonic::_codegen::HttpBody + Send + 'static,
<T::ResponseBody as tonic::_codegen::HttpBody>::Error: Into<tonic::error::Error> + Send,
<T::ResponseBody as tonic::_codegen::HttpBody>::Data: Send, {
<T::ResponseBody as tonic::_codegen::HttpBody>::Data: Into<bytes::Bytes> + 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
}
tonic-macros/src/service.rs
+2 -2
View File
@@ -106,7 +106,7 @@ pub(crate) fn generate(service: ServiceDef) -> TokenStream {
}
}
impl Service<http::Request<tower_h2::RecvBody>> for #server_service {
impl Service<http::Request<hyper::Body>> for #server_service {
type Response = http::Response<tonic::BoxBody>;
type Error = tonic::error::Never;
type Future = BoxFuture<Self::Response, Self::Error>;
@@ -115,7 +115,7 @@ pub(crate) fn generate(service: ServiceDef) -> TokenStream {
Poll::Ready(Ok(()))
}
fn call(&mut self, req: http::Request<tower_h2::RecvBody>) -> Self::Future {
fn call(&mut self, req: http::Request<hyper::Body>) -> Self::Future {
let inner = self.inner.clone();
match req.uri().path() {
tonic/Cargo.toml
+7 -13
View File
@@ -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" }
tonic/src/body.rs
+66 -6
View File
@@ -62,19 +62,33 @@ mod sealed {
}
pub struct BoxBody {
inner: Pin<Box<dyn HttpBody<Data = BytesBuf, Error = Status> + Send + 'static>>,
inner: Pin<Box<dyn Body<Data = BytesBuf, Error = Status> + Send + 'static>>,
}
struct MapBody<B>(B);
impl BoxBody {
/// Create a new `BoxBody` mapping item and error to the default types.
pub fn map_from<B>(inner: B) -> Self
pub fn new<B>(inner: B) -> Self
where
B: HttpBody<Data = BytesBuf, Error = Status> + Send + 'static,
B: Body<Data = BytesBuf, Error = Status> + Send + 'static,
{
BoxBody {
inner: Box::pin(inner),
}
}
/// Create a new `BoxBody` mapping item and error to the default types.
pub fn map_from<B>(inner: B) -> Self
where
B: Body + Send + 'static,
B::Data: Into<Bytes>,
B::Error: Into<crate::Error>,
{
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<Option<Result<Self::Data, Self::Error>>> {
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<Result<Option<http::HeaderMap>, Self::Error>> {
HttpBody::poll_trailers(self.inner.as_mut(), cx)
Body::poll_trailers(self.inner.as_mut(), cx)
}
}
impl<B> HttpBody for MapBody<B>
where
B: Body,
B::Data: Into<Bytes>,
B::Error: Into<crate::Error>,
{
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<Option<Result<Self::Data, Self::Error>>> {
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<Result<Option<http::HeaderMap>, 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)
}
}
tonic/src/client/grpc.rs
+34 -18
View File
@@ -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<T> Grpc<T> {
Self { inner }
}
pub async fn ready(&mut self) -> Result<(), Status>
where
T: GrpcService<BoxBody>,
T::ResponseBody: Body + HttpBody + Send + 'static,
<T::ResponseBody as HttpBody>::Error: Into<crate::Error> + Send,
<T::ResponseBody as HttpBody>::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<M1, M2, C>(
&mut self,
request: Request<M1>,
@@ -30,7 +48,7 @@ impl<T> Grpc<T> {
T: GrpcService<BoxBody>,
T::ResponseBody: Body + HttpBody + Send + 'static,
<T::ResponseBody as HttpBody>::Error: Into<crate::Error> + Send,
<T::ResponseBody as HttpBody>::Data: Send,
<T::ResponseBody as HttpBody>::Data: Into<Bytes> + Send,
C: Codec<Encode = M1, Decode = M2>,
C::Encoder: Send + 'static,
C::Decoder: Send + 'static,
@@ -51,7 +69,7 @@ impl<T> Grpc<T> {
T: GrpcService<BoxBody>,
T::ResponseBody: Body + HttpBody + Send + 'static,
<T::ResponseBody as HttpBody>::Error: Into<crate::Error> + Send,
<T::ResponseBody as HttpBody>::Data: Send,
<T::ResponseBody as HttpBody>::Data: Into<Bytes> + Send,
S: Stream<Item = Result<M1, Status>> + Send + 'static,
C: Codec<Encode = M1, Decode = M2>,
C::Encoder: Send + 'static,
@@ -59,7 +77,9 @@ impl<T> Grpc<T> {
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<T> Grpc<T> {
T: GrpcService<BoxBody>,
T::ResponseBody: Body + HttpBody + Send + 'static,
<T::ResponseBody as HttpBody>::Error: Into<crate::Error> + Send,
<T::ResponseBody as HttpBody>::Data: Send,
<T::ResponseBody as HttpBody>::Data: Into<Bytes> + Send,
C: Codec<Encode = M1, Decode = M2>,
C::Encoder: Send + 'static,
C::Decoder: Send + 'static,
@@ -100,7 +120,7 @@ impl<T> Grpc<T> {
T: GrpcService<BoxBody>,
T::ResponseBody: Body + HttpBody + Send + 'static,
<T::ResponseBody as HttpBody>::Error: Into<crate::Error> + Send,
<T::ResponseBody as HttpBody>::Data: Send,
<T::ResponseBody as HttpBody>::Data: Into<Bytes> + Send,
S: Stream<Item = Result<M1, Status>> + Send + 'static,
C: Codec<Encode = M1, Decode = M2>,
C::Encoder: Send + 'static,
@@ -115,7 +135,7 @@ impl<T> Grpc<T> {
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<T> Grpc<T> {
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))
}
tonic/src/codec/decode.rs
+167 -162
View File
@@ -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<T, B>(
decoder: T,
source: B,
) -> impl TryStream<Ok = T::Item, Error = Status> + 'static
where
T: Decoder<Error = Status> + 'static,
T::Item: Unpin + 'static,
B: Body + 'static,
B::Error: Into<crate::Error>,
{
decode(decoder, source, Direction::Request)
}
pub fn decode_response<T, B>(
decoder: T,
source: B,
status: StatusCode,
) -> impl TryStream<Ok = T::Item, Error = Status> + 'static
where
T: Decoder<Error = Status> + 'static,
T::Item: Unpin + 'static,
B: Body + 'static,
B::Error: Into<crate::Error>,
{
decode(decoder, source, Direction::Response(status))
}
pub fn decode_empty<T, B>(
decoder: T,
source: B,
) -> impl TryStream<Ok = T::Item, Error = Status> + 'static
where
T: Decoder<Error = Status> + 'static,
T::Item: Unpin + 'static,
B: Body + 'static,
B::Error: Into<crate::Error>,
{
decode(decoder, source, Direction::EmptyResponse)
}
// #[derive(Debug)]
pub struct Streaming<T> {
inner: Pin<Box<dyn Stream<Item = Result<T, Status>> + Send + 'static>>,
decoder: Box<dyn Decoder<Item = T, Error = Status> + Send + 'static>,
body: BoxBody,
state: State,
direction: Direction,
buf: BytesMut,
}
impl<T> Streaming<T> {
pub fn new(inner: impl Stream<Item = Result<T, Status>> + Send + 'static) -> Self {
let inner = Box::pin(inner);
Self { inner }
}
}
impl<T> Stream for Streaming<T> {
type Item = Result<T, Status>;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
Pin::new(&mut self.inner).poll_next(cx)
}
}
impl<T> fmt::Debug for Streaming<T> {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
write!(f, "Streaming")
}
}
impl<T> Unpin for Streaming<T> {}
#[derive(Debug)]
enum State {
@@ -88,50 +36,160 @@ enum Direction {
EmptyResponse,
}
fn decode<T, B>(
mut decoder: T,
mut source: B,
direction: Direction,
) -> impl TryStream<Ok = T::Item, Error = Status> + 'static
where
T: Decoder<Error = Status> + 'static,
T::Item: Unpin + 'static,
B: Body + 'static,
B::Error: Into<crate::Error>,
{
async_stream::try_stream! {
let mut buf = BytesMut::with_capacity(1024 * 1024 * 1024);
let mut state = State::ReadHeader;
impl<T> Streaming<T> {
pub fn new_response<B, D>(decoder: D, body: B, status_code: StatusCode) -> Self
where
B: Body + Send + 'static,
B::Data: Into<Bytes>,
B::Error: Into<crate::Error>,
D: Decoder<Item = T, Error = Status> + 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<B, D>(decoder: D, body: B) -> Self
where
B: Body + Send + 'static,
B::Data: Into<Bytes>,
B::Error: Into<crate::Error>,
D: Decoder<Item = T, Error = Status> + 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<B, D>(decoder: D, body: B) -> Self
where
B: Body + Send + 'static,
B::Data: Into<Bytes>,
B::Error: Into<crate::Error>,
D: Decoder<Item = T, Error = Status> + 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<T> Streaming<T> {
// pub async fn message(&mut self) -> Option<Result<T::Item, Status>> {
// future::poll_fn(|cx| Pin::new(&mut *self).poll_next(cx)).await
// }
pub async fn trailers(&mut self) -> Result<Option<MetadataMap>, 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<Option<T>, 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<T> Stream for Streaming<T> {
type Item = Result<T, Status>;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
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<T>(
decoder: &mut T,
buf1: &mut BytesMut,
state: &mut State,
) -> Result<Option<T::Item>, Status>
where
T: Decoder<Error = Status>,
{
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<T> fmt::Debug for Streaming<T> {
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)
}
tonic/src/codec/mod.rs
+1 -1
View File
@@ -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;
tonic/src/lib.rs
+2 -2
View File
@@ -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<Request> {
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;
tonic/src/server/grpc.rs
+12 -13
View File
@@ -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<T::Decode, Response = T::Encode>,
B: Body + Send + 'static,
B::Data: Send,
B::Data: Into<Bytes> + Send,
B::Error: Into<crate::Error> + Send,
{
let request = match self.map_request_unary(req).await {
@@ -62,7 +63,7 @@ where
S: ServerStreamingService<T::Decode, Response = T::Encode>,
S::ResponseStream: Send + 'static,
B: Body + Send + 'static,
B::Data: Send,
B::Data: Into<Bytes> + Send,
B::Error: Into<crate::Error> + 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<Bytes> + Send + 'static,
B::Error: Into<crate::Error> + Send + 'static,
{
let request = self.map_request_streaming(req);
@@ -108,7 +109,7 @@ where
S: StreamingService<Streaming<T::Decode>, Response = T::Encode> + Send,
S::ResponseStream: Send + 'static,
B: Body + Send + 'static,
B::Data: Send,
B::Data: Into<Bytes> + Send,
B::Error: Into<crate::Error> + Send,
{
let request = self.map_request_streaming(req);
@@ -122,11 +123,11 @@ where
) -> Result<Request<T::Decode>, Status>
where
B: Body + Send + 'static,
B::Data: Send,
B::Data: Into<Bytes> + Send,
B::Error: Into<crate::Error> + 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<Streaming<T::Decode>>
where
B: Body + Send + 'static,
B::Data: Send,
B::Data: Into<Bytes> + Send,
B::Error: Into<crate::Error> + 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<B>(
@@ -173,7 +172,7 @@ where
// FIXME: try to return impl Trait?
// let body = Box::pin(body) as BoxStream<BytesBuf>;
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))
}
}
}
tower-h2/src/add_origin.rs → tonic/src/service/add_origin.rs
View File
tonic/src/service.rs → tonic/src/service/mod.rs
+2
View File
@@ -1,3 +1,5 @@
pub mod add_origin;
use crate::body::Body;
use http::{Request, Response};
use http_body::Body as HttpBody;
tower-h2/Cargo.toml
-26
View File
@@ -1,26 +0,0 @@
[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" }
# 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"
tower-h2/examples/client.rs
-57
View File
@@ -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<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 http_body::Body for Body {
type Data = std::io::Cursor<Vec<u8>>;
type Error = std::io::Error;
fn poll_data(
mut self: Pin<&mut Self>,
_cx: &mut Context<'_>,
) -> Poll<Option<Result<Self::Data, 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()
}
fn poll_trailers(
self: Pin<&mut Self>,
_cx: &mut Context<'_>,
) -> Poll<Result<Option<http::HeaderMap>, Self::Error>> {
Ok(None).into()
}
}
tower-h2/examples/server.rs
-111
View File
@@ -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<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 http_body::Body for Body {
type Data = std::io::Cursor<Vec<u8>>;
type Error = std::io::Error;
fn poll_data(
mut self: Pin<&mut Self>,
_cx: &mut Context<'_>,
) -> Poll<Option<Result<Self::Data, 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()
}
fn poll_trailers(
self: Pin<&mut Self>,
_cx: &mut Context<'_>,
) -> Poll<Result<Option<http::HeaderMap>, Self::Error>> {
Ok(None).into()
}
}
tower-h2/src/buf.rs
-38
View File
@@ -1,38 +0,0 @@
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
-81
View File
@@ -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<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, 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)))
}
}
tower-h2/src/error.rs
-12
View File
@@ -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::<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
-196
View File
@@ -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<S>
where
S: Body,
{
h2: SendStream<SendBuf<S::Data>>,
body: Pin<Box<dyn Body<Data = S::Data, Error = S::Error> + Send + 'static>>,
state: FlushState,
}
#[derive(Debug)]
enum FlushState {
Data,
Trailers,
Done,
}
enum DataOrTrailers<B> {
Data(B),
Trailers(HeaderMap),
}
// ===== impl Flush =====
impl<S> Flush<S>
where
S: Body + Send + 'static,
S::Error: Into<Box<dyn std::error::Error>>,
{
pub fn new(src: S, dst: SendStream<SendBuf<S::Data>>) -> 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<Result<(), h2::Error>> {
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<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 {
// 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<S> Future for Flush<S>
where
S: Body + Send + 'static,
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
-15
View File
@@ -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};
tower-h2/src/recv_body.rs
-98
View File
@@ -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<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: Pin<&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
-110
View File
@@ -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<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, 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);
}
}
}