diff --git a/tonic-examples/Cargo.toml b/tonic-examples/Cargo.toml index 8517db8..453a180 100644 --- a/tonic-examples/Cargo.toml +++ b/tonic-examples/Cargo.toml @@ -29,7 +29,7 @@ prost-derive = "0.5" bytes = "0.4" serde_json = "1.0" serde = { version = "1.0", features = ["derive"] } -async-stream = { git = "https://github.com/tokio-rs/async-stream" } +async-stream = "0.1.1" http = "0.1" [build-dependencies] diff --git a/tonic-interop/Cargo.toml b/tonic-interop/Cargo.toml index 63a92aa..7c2c6c2 100644 --- a/tonic-interop/Cargo.toml +++ b/tonic-interop/Cargo.toml @@ -21,7 +21,7 @@ bytes = "0.4" http = "0.1" futures-core-preview = "=0.3.0-alpha.18" futures-util-preview = "=0.3.0-alpha.18" -async-stream = { git = "https://github.com/tokio-rs/async-stream" } +async-stream = "0.1.1" console = "0.7" structopt = "0.2" diff --git a/tonic-interop/test.sh b/tonic-interop/test.sh index 3ec5ed6..aa79ce3 100755 --- a/tonic-interop/test.sh +++ b/tonic-interop/test.sh @@ -9,6 +9,8 @@ case "$OSTYPE" in *) exit 2 ;; esac +cargo build -p tonic-interop --bins + ARG="${1:-""}" SERVER="tonic-interop/bin/server_${OS}_amd64" @@ -25,7 +27,7 @@ echo ":; started grpc-go test server." # regardless of why (errors, SIGTERM, etc). trap 'echo ":; killing test server"; kill ${SERVER_PID};' EXIT - cargo run -p tonic-interop --bin client -- \ +./target/debug/client \ --test_case=empty_unary,large_unary,client_streaming,server_streaming,ping_pong,\ empty_stream,status_code_and_message,special_status_message,unimplemented_method,\ unimplemented_service,custom_metadata $ARG @@ -41,7 +43,7 @@ echo ":; started tonic test server." # regardless of why (errors, SIGTERM, etc). trap 'echo ":; killing test server"; kill ${SERVER_PID};' EXIT -cargo run -p tonic-interop --bin client -- \ +./target/debug/client \ --test_case=empty_unary,large_unary,client_streaming,server_streaming,ping_pong,\ empty_stream,status_code_and_message,special_status_message $ARG # unimplemented_method,unimplemented_service,custom_metadata diff --git a/tonic/Cargo.toml b/tonic/Cargo.toml index d5ae6fa..393d56a 100644 --- a/tonic/Cargo.toml +++ b/tonic/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "tonic" -version = "0.1.0" +version = "0.1.0-alpha.1" authors = ["Lucio Franco "] edition = "2018" @@ -15,20 +15,21 @@ prost = "0.5" percent-encoding = "1.0.1" tower-service = "=0.3.0-alpha.1" tokio-codec = "=0.2.0-alpha.4" -async-stream = { git = "https://github.com/tokio-rs/async-stream" } +async-stream = "0.1.1" http-body = "0.2.0-alpha.1" -pin-project = "=0.4.0-alpha.7" +pin-project = "=0.4.0-alpha.11" async-trait = "0.1" # optional -hyper = { version = "=0.13.0-alpha.1", optional = true } +# hyper = { version = "=0.13.0-alpha.1", optional = true } +hyper = { git = "https://github.com/hyperium/hyper", features = ["unstable-stream"], optional = true } tokio = { version = "=0.2.0-alpha.4", default-features = false, features = ["tcp"], optional = true } tower-make = "=0.1.0-alpha.2" -tower-reconnect = { git = "https://github.com/tower-rs/tower", branch = "lucio/update-reconnect-buffer", optional = true } -tower-buffer = { git = "https://github.com/tower-rs/tower", branch = "lucio/update-reconnect-buffer", optional = true } -tower-balance = { git = "https://github.com/tower-rs/tower", branch = "lucio/update-balance", optional = true } -tower-load = { git = "https://github.com/tower-rs/tower", branch = "lucio/update-balance", optional = true } -tower-discover = { git = "https://github.com/tower-rs/tower", branch = "lucio/update-balance", optional = true } +tower-reconnect = { version = "0.3.0-alpha.1", optional = true } +tower-buffer = { version = "0.3.0-alpha.1", optional = true } +tower-balance = { version = "0.3.0-alpha.1", optional = true } +tower-load = { version = "0.3.0-alpha.1", optional = true } +tower-discover = { version = "0.3.0-alpha.1", optional = true } # openssl tokio-openssl = { version = "=0.4.0-alpha.4", optional = true } diff --git a/tonic/src/body.rs b/tonic/src/body.rs index f793efb..41930ed 100644 --- a/tonic/src/body.rs +++ b/tonic/src/body.rs @@ -210,6 +210,7 @@ impl HttpBody for EmptyBody { self: Pin<&mut Self>, cx: &mut Context<'_>, ) -> Poll>> { + drop(cx); Poll::Ready(None) } @@ -217,6 +218,7 @@ impl HttpBody for EmptyBody { self: Pin<&mut Self>, cx: &mut Context<'_>, ) -> Poll, Self::Error>> { + drop(cx); Poll::Ready(Ok(None)) } } diff --git a/tonic/src/codec/decode.rs b/tonic/src/codec/decode.rs index 2ec7c53..7c72b0b 100644 --- a/tonic/src/codec/decode.rs +++ b/tonic/src/codec/decode.rs @@ -112,7 +112,7 @@ impl Streaming { } // To fetch the trailers we must clear the body and drop it. - while let Some(res) = self.message().await? {} + while let Some(_) = self.message().await? {} // Since we call poll_trailers internally on poll_next we need to // check if it got cached again. diff --git a/tonic/src/lib.rs b/tonic/src/lib.rs index afb1088..6241f24 100644 --- a/tonic/src/lib.rs +++ b/tonic/src/lib.rs @@ -1,10 +1,12 @@ -#![recursion_limit = "512"] +#![recursion_limit = "256"] #![warn( missing_debug_implementations, missing_docs, rust_2018_idioms, unreachable_pub )] +#![doc(html_logo_url = "file:///Users/lucio/Downloads/tonic_bubbles_with_word_bigger.svg")] +#![doc(html_root_url = "https://docs.rs/tonic/0.1.0")] #![doc(test(no_crate_inject, attr(deny(rust_2018_idioms))))] //! A rust implementation of [gRPC], a high performance, open source, general diff --git a/tonic/src/transport/channel.rs b/tonic/src/transport/channel.rs index 01aab5c..d45ef0b 100644 --- a/tonic/src/transport/channel.rs +++ b/tonic/src/transport/channel.rs @@ -89,7 +89,7 @@ impl Builder { fn balance(&mut self, discover: D) -> Result where - D: Discover + Send + 'static, + D: Discover + Unpin + Send + 'static, D::Error: Into, D::Key: Send + Clone, { diff --git a/tonic/src/transport/server.rs b/tonic/src/transport/server.rs index dca6eea..43e40e6 100644 --- a/tonic/src/transport/server.rs +++ b/tonic/src/transport/server.rs @@ -3,12 +3,12 @@ use super::{ tls::{Cert, TlsAcceptor}, }; use crate::BoxBody; -use futures_util::{try_future::MapOk, TryFutureExt, TryStreamExt}; +use futures_core::Stream; +use futures_util::{try_future::MapOk, TryFutureExt, TryStreamExt, ready}; use http::{Request, Response}; -use hyper::server::conn; +use hyper::server::{conn, accept::Accept}; use hyper::Body; -use std::net::SocketAddr; -use std::task::{Context, Poll}; +use std::{net::SocketAddr, task::{Context, Poll}, pin::Pin}; use tower_make::MakeService; use tower_service::Service; @@ -48,8 +48,6 @@ impl Builder { S::Future: Send + 'static, S::Error: Into, { - let tcp = conn::AddrIncoming::bind(&addr).unwrap(); - let tls = if let Some(tls) = self.tls { let cert = Cert { ca: tls.0, @@ -62,7 +60,7 @@ impl Builder { None }; - let incoming = incoming(tcp, tls); + let incoming = hyper::server::accept::from_stream(incoming(addr, tls)); let svc = MakeSvc(svc); @@ -77,11 +75,13 @@ impl Builder { } fn incoming( - mut tcp: conn::AddrIncoming, + addr: SocketAddr, tls: Option, ) -> impl futures_core::Stream> { async_stream::try_stream! { - while let Some(stream) = tcp.try_next().await.map_err(Into::into)? { + let mut tcp = TcpIncoming::bind(addr)?; + + while let Some(stream) = tcp.try_next().await? { if let Some(tls) = &tls { let io = tls.connect(stream.into_inner()).await?; yield BoxedIo::new(io); @@ -92,9 +92,36 @@ fn incoming( } } +#[derive(Debug)] +struct TcpIncoming { + inner: conn::AddrIncoming, +} + +impl TcpIncoming { + fn bind(addr: SocketAddr) -> Result { + let inner = conn::AddrIncoming::bind(&addr).map_err(Box::new)?; + + Ok(Self { + inner, + }) + } +} + +impl Stream for TcpIncoming { + type Item = Result; + + fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + match ready!(Accept::poll_accept(Pin::new(&mut self.inner), cx)) { + Some(Ok(s)) => Poll::Ready(Some(Ok(s))), + Some(Err(e)) => Poll::Ready(Some(Err(e.into()))), + None => Poll::Ready(None), + } + } +} + // TODO: add custom tracing here #[derive(Debug)] -pub(crate) struct Svc(S); +struct Svc(S); impl Service> for Svc where @@ -113,7 +140,7 @@ where } } -pub(crate) struct MakeSvc(M); +struct MakeSvc(M); impl Service for MakeSvc where diff --git a/tonic/src/transport/service/discover.rs b/tonic/src/transport/service/discover.rs index 9756096..a59c8ae 100644 --- a/tonic/src/transport/service/discover.rs +++ b/tonic/src/transport/service/discover.rs @@ -1,6 +1,7 @@ use super::connect::Connection; use crate::transport::Endpoint; use std::collections::VecDeque; +use std::pin::Pin; use std::task::{Context, Poll}; use tower_discover::{Change, Discover}; @@ -24,8 +25,8 @@ impl Discover for ServiceList { type Service = Connection; type Error = crate::Error; - fn poll( - &mut self, + fn poll_discover( + mut self: Pin<&mut Self>, _cx: &mut Context<'_>, ) -> Poll, Self::Error>> { match self.list.pop_front() { diff --git a/tonic/src/transport/tls/mod.rs b/tonic/src/transport/tls/mod.rs index 489adf2..2de644c 100644 --- a/tonic/src/transport/tls/mod.rs +++ b/tonic/src/transport/tls/mod.rs @@ -10,6 +10,7 @@ mod imp; use tokio::net::TcpStream; +use std::fmt; #[derive(Debug, Clone)] pub(crate) struct Cert { @@ -34,6 +35,12 @@ impl TlsConnector { } } +impl fmt::Debug for TlsConnector { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("TlsConnector").finish() + } +} + #[derive(Clone)] pub(crate) struct TlsAcceptor { inner: imp::TlsAcceptor, @@ -49,3 +56,9 @@ impl TlsAcceptor { self.inner.connect(io).await } } + +impl fmt::Debug for TlsAcceptor { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("TlsAcceptor").finish() + } +}