Upgrade to tokio 0.2 (hyperium#163)

Cargo.toml

@@ -9,3 +9,8 @@ members = [
     "tests/same_name",
     "tests/wellknown",
 ]
+
+[patch.'https://github.com/tower-rs/tower']
+tower-service = "0.3"
+tower-make = "0.3"
+tower-layer = "0.3"

deny.toml

@@ -15,8 +15,8 @@ deny = [
     { name = "term" },
 ]
 skip = [
-    { name = "crossbeam-utils", version = "=0.6.6" },
     { name = "crossbeam-queue", version = "=0.2.0" },
+    { name = "bytes", version = "=0.4.12" },
 ]
 skip-tree = [
     { name = "rand", version = "=0.6.5" },

tests/wellknown/Cargo.toml

@@ -10,7 +10,7 @@ license = "MIT"
 
 [dependencies]
 tonic = { path = "../../tonic" }
-bytes = "0.4"
+bytes = "0.5"
 prost = "0.5"
 prost-types = "0.5"
 

tonic-build/src/client.rs

@@ -23,7 +23,7 @@ pub(crate) fn generate(service: &Service, proto: &str) -> TokenStream {
               T::ResponseBody: Body + HttpBody + Send + 'static,
               T::Error: Into<StdError>,
               <T::ResponseBody as HttpBody>::Error: Into<StdError> + Send,
-              <T::ResponseBody as HttpBody>::Data: Into<bytes::Bytes> + Send, {
+        {
             pub fn new(inner: T) -> Self {
                 let inner = tonic::client::Grpc::new(inner);
                 Self { inner }

tonic-examples/Cargo.toml

@@ -71,11 +71,11 @@ tonic = { path = "../tonic", features = ["tls"] }
 bytes = "0.4"
 prost = "0.5"
 
-tokio = "=0.2.0-alpha.6"
-futures-preview = { version = "=0.3.0-alpha.19", default-features = false, features = ["alloc"]}
-async-stream = "0.1.2"
-http = "0.1"
-tower = "=0.3.0-alpha.2"
+tokio = { version = "0.2", features = ["rt-threaded", "time", "stream", "fs"] }
+futures = { version = "0.3", default-features = false, features = ["alloc"]}
+async-stream = "0.2"
+http = "0.2"
+tower = { git = "https://github.com/tower-rs/tower" }
 
 # Required for routeguide
 serde = { version = "1.0", features = ["derive"] }

tonic-examples/src/load_balance/server.rs

@@ -60,7 +60,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     for addr in &addrs {
         let addr = addr.parse()?;
-        let mut tx = tx.clone();
+        let tx = tx.clone();
 
         let server = EchoServer { addr };
         let serve = Server::builder()
@@ -72,7 +72,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
                 eprintln!("Error = {:?}", e);
             }
 
-            tx.try_send(()).unwrap();
+            tx.send(()).unwrap();
         });
     }
 

tonic-examples/src/routeguide/client.rs

@@ -3,8 +3,8 @@ use rand::rngs::ThreadRng;
 use rand::Rng;
 use route_guide::{Point, Rectangle, RouteNote};
 use std::error::Error;
-use std::time::{Duration, Instant};
-use tokio::timer::Interval;
+use std::time::Duration;
+use tokio::time::Instant;
 use tonic::transport::Channel;
 use tonic::Request;
 
@@ -62,9 +62,9 @@ async fn run_route_chat(client: &mut RouteGuideClient<Channel>) -> Result<(), Bo
     let start = Instant::now();
 
     let outbound = async_stream::stream! {
-        let mut interval = Interval::new_interval(Duration::from_secs(1));
+        let mut interval = tokio::time::interval(Duration::from_secs(1));
 
-        while let Some(time) = interval.next().await {
+        while let time = interval.tick().await {
             let elapsed = time.duration_since(start);
             let note = RouteNote {
                 location: Some(Point {

tonic-interop/Cargo.toml

@@ -15,17 +15,18 @@ name = "server"
 path = "src/bin/server.rs"
 
 [dependencies]
-tokio = "=0.2.0-alpha.6"
+tokio = { version = "0.2", features = ["rt-threaded", "time", "macros", "stream", "fs"] }
 tonic = { path = "../tonic", features = ["tls"] }
 prost = "0.5"
 prost-derive = "0.5"
 bytes = "0.4"
-http = "0.1"
-futures-core-preview = "=0.3.0-alpha.19"
-futures-util-preview = "=0.3.0-alpha.19"
-async-stream = "0.1.2"
-tower = "=0.3.0-alpha.2"
-http-body = "=0.2.0-alpha.3"
+http = "0.2"
+futures-core = "0.3"
+futures-util = "0.3"
+async-stream = "0.2"
+# tower = "=0.3.0-alpha.2"
+tower = { git = "https://github.com/tower-rs/tower" }
+http-body = "0.3"
 
 console = "0.9"
 structopt = "0.2"

tonic-interop/src/client.rs

@@ -1,5 +1,5 @@
 use crate::{pb::client::*, pb::*, test_assert, TestAssertion};
-use futures_util::{future, stream, SinkExt, StreamExt};
+use futures_util::{future, stream, StreamExt};
 use tokio::sync::mpsc;
 use tonic::transport::Channel;
 use tonic::{metadata::MetadataValue, Code, Request, Response, Status};
@@ -148,8 +148,8 @@ pub async fn server_streaming(client: &mut TestClient, assertions: &mut Vec<Test
 }
 
 pub async fn ping_pong(client: &mut TestClient, assertions: &mut Vec<TestAssertion>) {
-    let (mut tx, rx) = mpsc::unbounded_channel();
-    tx.try_send(make_ping_pong_request(0)).unwrap();
+    let (tx, rx) = mpsc::unbounded_channel();
+    tx.send(make_ping_pong_request(0)).unwrap();
 
     let result = client.full_duplex_call(Request::new(rx)).await;
 
@@ -170,9 +170,7 @@ pub async fn ping_pong(client: &mut TestClient, assertions: &mut Vec<TestAsserti
                         drop(tx);
                         break;
                     } else {
-                        tx.send(make_ping_pong_request(responses.len()))
-                            .await
-                            .unwrap();
+                        tx.send(make_ping_pong_request(responses.len())).unwrap();
                     }
                 }
                 None => {

tonic-interop/src/server.rs

@@ -2,7 +2,7 @@ use crate::pb::{self, *};
 use async_stream::try_stream;
 use futures_util::{stream, StreamExt, TryStreamExt};
 use std::pin::Pin;
-use std::time::{Duration, Instant};
+use std::time::Duration;
 use tonic::{Code, Request, Response, Status};
 
 pub use pb::server::{TestServiceServer, UnimplementedServiceServer};
@@ -65,8 +65,7 @@ impl pb::server::TestService for TestService {
 
         let stream = try_stream! {
             for param in response_parameters {
-                let deadline = Instant::now() + Duration::from_micros(param.interval_us as u64);
-                tokio::timer::delay(deadline).await;
+                tokio::time::delay_for(Duration::from_micros(param.interval_us as u64)).await;
 
                 let payload = crate::server_payload(param.size as usize);
                 yield StreamingOutputCallResponse { payload: Some(payload) };
@@ -121,8 +120,7 @@ impl pb::server::TestService for TestService {
                     }
 
                     for param in msg.response_parameters {
-                        let deadline = Instant::now() + Duration::from_micros(param.interval_us as u64);
-                        tokio::timer::delay(deadline).await;
+                        tokio::time::delay_for(Duration::from_micros(param.interval_us as u64)).await;
 
                         let payload = crate::server_payload(param.size as usize);
                         yield StreamingOutputCallResponse { payload: Some(payload) };

tonic/Cargo.toml

@@ -35,24 +35,24 @@ transport = [
 tls = ["tokio-rustls"]
 tls-roots = ["rustls-native-certs"]
 
-[[bench]]
-name = "bench_main"
-harness = false
+# [[bench]]
+# name = "bench_main"
+# harness = false
 
 [dependencies]
-bytes = "0.4"
-futures-core-preview = "=0.3.0-alpha.19"
-futures-util-preview = { version = "=0.3.0-alpha.19", default-features = false }
+bytes = "0.5"
+futures-core = "0.3"
+futures-util = { version = "0.3", default-features = false }
 tracing = "0.1"
-http = "0.1.14"
+http = "0.2"
 base64 = "0.10"
 
 percent-encoding = "1.0.1"
-tower-service = "=0.3.0-alpha.2"
-tokio-codec = "=0.2.0-alpha.6"
-async-stream = "0.1.2"
-http-body = "=0.2.0-alpha.3"
-pin-project = "^0.4"
+tower-service = "0.3"
+tokio-util = { version = "0.2", features = ["codec"] }
+async-stream = "0.2"
+http-body = "0.3"
+pin-project = "0.4"
 
 # prost
 prost = { version = "0.5", optional = true }
@@ -62,22 +62,25 @@ prost-derive = { version = "0.5", optional = true }
 async-trait = { version = "0.1.13", optional = true }
 
 # transport
-hyper = { version = "=0.13.0-alpha.4", features = ["unstable-stream"], optional = true }
-tokio = { version = "=0.2.0-alpha.6", default-features = false, features = ["tcp"], optional = true }
-tower = { version = "=0.3.0-alpha.2", optional = true}
-tower-make = "=0.3.0-alpha.2a"
-tower-balance =  { version = "=0.3.0-alpha.2", optional = true }
-tower-load = { version = "=0.3.0-alpha.2", optional = true }
+hyper = { git = "https://github.com/hyperium/hyper", features = ["stream"], optional = true }
+tokio = { version = "0.2", features = ["tcp"], optional = true }
+tower = { git = "https://github.com/tower-rs/tower", optional = true}
+tower-make = { version = "0.3", features = ["connect"] }
+tower-balance =  { git = "https://github.com/tower-rs/tower", optional = true }
+tower-load = { git = "https://github.com/tower-rs/tower", optional = true }
 
 # rustls
-tokio-rustls = { version = "=0.12.0-alpha.5", optional = true }
+tokio-rustls = { version = "0.12", optional = true }
 rustls-native-certs = { version = "0.1", optional = true }
 
 [dev-dependencies]
+tokio = { version = "0.2", features = ["rt-core", "macros"] }
 static_assertions = "1.0"
 rand = "0.7.2"
 criterion = "0.3"
 
 [package.metadata.docs.rs]
 all-features = true
 rustdoc-args = ["--cfg", "docsrs"]
+
+

tonic/benches/bench_main.rs

@@ -1,3 +1,5 @@
+#![cfg(feature = "broken")]
+
 use criterion::*;
 
 mod benchmarks;

tonic/src/body.rs

@@ -4,16 +4,14 @@
 //! of the types in this module are based around [`http_body::Body`].
 
 use crate::{Error, Status};
-use bytes::{Buf, Bytes, IntoBuf};
+use bytes::{Buf, Bytes};
 use http_body::Body as HttpBody;
 use std::{
     fmt,
     pin::Pin,
     task::{Context, Poll},
 };
 
-pub(crate) type BytesBuf = <Bytes as IntoBuf>::Buf;
-
 /// A trait alias for [`http_body::Body`].
 pub trait Body: sealed::Sealed + Send + Sync {
     /// The body data type.
@@ -83,7 +81,7 @@ mod sealed {
 
 /// A type erased http body.
 pub struct BoxBody {
-    inner: Pin<Box<dyn Body<Data = BytesBuf, Error = Status> + Send + Sync + 'static>>,
+    inner: Pin<Box<dyn Body<Data = Bytes, Error = Status> + Send + Sync + 'static>>,
 }
 
 struct MapBody<B>(B);
@@ -92,7 +90,7 @@ impl BoxBody {
     /// Create a new `BoxBody` mapping item and error to the default types.
     pub fn new<B>(inner: B) -> Self
     where
-        B: Body<Data = BytesBuf, Error = Status> + Send + Sync + 'static,
+        B: Body<Data = Bytes, Error = Status> + Send + Sync + 'static,
     {
         BoxBody {
             inner: Box::pin(inner),
@@ -103,7 +101,7 @@ impl BoxBody {
     pub fn map_from<B>(inner: B) -> Self
     where
         B: Body + Send + Sync + 'static,
-        B::Data: Into<Bytes>,
+        // B::Data: Into<Bytes>,
         B::Error: Into<crate::Error>,
     {
         BoxBody {
@@ -120,7 +118,7 @@ impl BoxBody {
 }
 
 impl HttpBody for BoxBody {
-    type Data = BytesBuf;
+    type Data = Bytes;
     type Error = Status;
 
     fn is_end_stream(&self) -> bool {
@@ -145,10 +143,10 @@ impl HttpBody for BoxBody {
 impl<B> HttpBody for MapBody<B>
 where
     B: Body,
-    B::Data: Into<Bytes>,
+    // B::Data: Into<Bytes>,
     B::Error: Into<crate::Error>,
 {
-    type Data = BytesBuf;
+    type Data = Bytes;
     type Error = Status;
 
     fn is_end_stream(&self) -> bool {
@@ -164,7 +162,7 @@ where
             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(Ok(mut i)) => Poll::Ready(Some(Ok(i.to_bytes()))),
             Some(Err(e)) => {
                 let err = Status::map_error(e.into());
                 Poll::Ready(Some(Err(err)))
@@ -199,7 +197,7 @@ struct EmptyBody {
 }
 
 impl HttpBody for EmptyBody {
-    type Data = BytesBuf;
+    type Data = Bytes;
     type Error = Status;
 
     fn is_end_stream(&self) -> bool {

tonic/src/client/grpc.rs

@@ -4,7 +4,6 @@ use crate::{
     codec::{encode_client, Codec, Streaming},
     Code, Request, Response, Status,
 };
-use bytes::Bytes;
 use futures_core::Stream;
 use futures_util::{future, stream, TryStreamExt};
 use http::{
@@ -60,7 +59,7 @@ impl<T> Grpc<T> {
         T: GrpcService<BoxBody>,
         T::ResponseBody: Body + HttpBody + Send + 'static,
         <T::ResponseBody as HttpBody>::Error: Into<crate::Error>,
-        <T::ResponseBody as HttpBody>::Data: Into<Bytes>,
+        // <T::ResponseBody as HttpBody>::Data: Into<Bytes>,
         C: Codec<Encode = M1, Decode = M2>,
         M1: Send + Sync + 'static,
         M2: Send + Sync + 'static,
@@ -80,7 +79,7 @@ impl<T> Grpc<T> {
         T: GrpcService<BoxBody>,
         T::ResponseBody: Body + HttpBody + Send + 'static,
         <T::ResponseBody as HttpBody>::Error: Into<crate::Error>,
-        <T::ResponseBody as HttpBody>::Data: Into<Bytes>,
+        // <T::ResponseBody as HttpBody>::Data: Into<Bytes>,
         S: Stream<Item = M1> + Send + Sync + 'static,
         C: Codec<Encode = M1, Decode = M2>,
         M1: Send + Sync + 'static,
@@ -113,7 +112,7 @@ impl<T> Grpc<T> {
         T: GrpcService<BoxBody>,
         T::ResponseBody: Body + HttpBody + Send + 'static,
         <T::ResponseBody as HttpBody>::Error: Into<crate::Error>,
-        <T::ResponseBody as HttpBody>::Data: Into<Bytes>,
+        // <T::ResponseBody as HttpBody>::Data: Into<Bytes>,
         C: Codec<Encode = M1, Decode = M2>,
         M1: Send + Sync + 'static,
         M2: Send + Sync + 'static,
@@ -132,7 +131,7 @@ impl<T> Grpc<T> {
     where
         T: GrpcService<BoxBody>,
         T::ResponseBody: Body + HttpBody + Send + 'static,
-        <T::ResponseBody as HttpBody>::Data: Into<Bytes>,
+        // <T::ResponseBody as HttpBody>::Data: Into<Bytes>,
         <T::ResponseBody as HttpBody>::Error: Into<crate::Error>,
         S: Stream<Item = M1> + Send + Sync + 'static,
         C: Codec<Encode = M1, Decode = M2>,