Update deps and tower releases

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]

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"

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

tonic/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "tonic"
-version = "0.1.0"
+version = "0.1.0-alpha.1"
 authors = ["Lucio Franco <[email protected]>"]
 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-reconnect = { version = "0.3.0-alpha.1", optional = true }
-tower-buffer = { git = "https://github.com/tower-rs/tower", branch = "lucio/update-reconnect-buffer", optional = true }
+tower-buffer = { version = "0.3.0-alpha.1", optional = true }
-tower-balance = { git = "https://github.com/tower-rs/tower", branch = "lucio/update-balance", optional = true }
+tower-balance =  { version = "0.3.0-alpha.1", optional = true }
-tower-load = { git = "https://github.com/tower-rs/tower", branch = "lucio/update-balance", optional = true }
+tower-load = { version = "0.3.0-alpha.1", optional = true }
-tower-discover = { git = "https://github.com/tower-rs/tower", branch = "lucio/update-balance", optional = true }
+tower-discover = { version = "0.3.0-alpha.1", optional = true }
 
 # openssl
 tokio-openssl = { version = "=0.4.0-alpha.4", optional = true }

tonic/src/body.rs

@@ -210,6 +210,7 @@ impl HttpBody for EmptyBody {
         self: Pin<&mut Self>,
         cx: &mut Context<'_>,
     ) -> Poll<Option<Result<Self::Data, Self::Error>>> {
+        drop(cx);
         Poll::Ready(None)
     }
 
@@ -217,6 +218,7 @@ impl HttpBody for EmptyBody {
         self: Pin<&mut Self>,
         cx: &mut Context<'_>,
     ) -> Poll<Result<Option<http::HeaderMap>, Self::Error>> {
+        drop(cx);
         Poll::Ready(Ok(None))
     }
 }

tonic/src/codec/decode.rs

@@ -112,7 +112,7 @@ impl<T> Streaming<T> {
         }
 
         // 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.

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

tonic/src/transport/channel.rs

@@ -89,7 +89,7 @@ impl Builder {
 
     fn balance<D>(&mut self, discover: D) -> Result<Channel, super::Error>
     where
-        D: Discover<Service = Connection> + Send + 'static,
+        D: Discover<Service = Connection> + Unpin + Send + 'static,
         D::Error: Into<crate::Error>,
         D::Key: Send + Clone,
     {

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::{net::SocketAddr, task::{Context, Poll}, pin::Pin};
-use std::task::{Context, Poll};
 use tower_make::MakeService;
 use tower_service::Service;
 
@@ -48,8 +48,6 @@ impl Builder {
         S::Future: Send + 'static,
         S::Error: Into<crate::Error>,
     {
-        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<TlsAcceptor>,
 ) -> impl futures_core::Stream<Item = Result<BoxedIo, crate::Error>> {
     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<Self, crate::Error> {
+        let inner = conn::AddrIncoming::bind(&addr).map_err(Box::new)?;
+
+        Ok(Self {
+            inner,
+        })
+    }
+}
+
+impl Stream for TcpIncoming {
+    type Item = Result<conn::AddrStream, crate::Error>;
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        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>(S);
+struct Svc<S>(S);
 
 impl<S> Service<Request<Body>> for Svc<S>
 where
@@ -113,7 +140,7 @@ where
     }
 }
 
-pub(crate) struct MakeSvc<M>(M);
+struct MakeSvc<M>(M);
 
 impl<M, S, T> Service<T> for MakeSvc<M>
 where

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(
+    fn poll_discover(
-        &mut self,
+        mut self: Pin<&mut Self>,
         _cx: &mut Context<'_>,
     ) -> Poll<Result<Change<Self::Key, Self::Service>, Self::Error>> {
         match self.list.pop_front() {

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()
+    }
+}