Upgrade to tokio 0.2 (#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"
+tokio = { version = "0.2", features = ["rt-threaded", "time", "stream", "fs"] }
-futures-preview = { version = "=0.3.0-alpha.19", default-features = false, features = ["alloc"]}
+futures = { version = "0.3", default-features = false, features = ["alloc"]}
-async-stream = "0.1.2"
+async-stream = "0.2"
-http = "0.1"
+http = "0.2"
-tower = "=0.3.0-alpha.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 std::time::Duration;
-use tokio::timer::Interval;
+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"
+http = "0.2"
-futures-core-preview = "=0.3.0-alpha.19"
+futures-core = "0.3"
-futures-util-preview = "=0.3.0-alpha.19"
+futures-util = "0.3"
-async-stream = "0.1.2"
+async-stream = "0.2"
-tower = "=0.3.0-alpha.2"
+# tower = "=0.3.0-alpha.2"
-http-body = "=0.2.0-alpha.3"
+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();
+    let (tx, rx) = mpsc::unbounded_channel();
-    tx.try_send(make_ping_pong_request(0)).unwrap();
+    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()))
+                        tx.send(make_ping_pong_request(responses.len())).unwrap();
-                            .await
-                            .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::time::delay_for(Duration::from_micros(param.interval_us as u64)).await;
-                tokio::timer::delay(deadline).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::time::delay_for(Duration::from_micros(param.interval_us as u64)).await;
-                        tokio::timer::delay(deadline).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]]
+# [[bench]]
-name = "bench_main"
+# name = "bench_main"
-harness = false
+# harness = false
 
 [dependencies]
-bytes = "0.4"
+bytes = "0.5"
-futures-core-preview = "=0.3.0-alpha.19"
+futures-core = "0.3"
-futures-util-preview = { version = "=0.3.0-alpha.19", default-features = false }
+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"
+tower-service = "0.3"
-tokio-codec = "=0.2.0-alpha.6"
+tokio-util = { version = "0.2", features = ["codec"] }
-async-stream = "0.1.2"
+async-stream = "0.2"
-http-body = "=0.2.0-alpha.3"
+http-body = "0.3"
-pin-project = "^0.4"
+pin-project = "0.4"
 
 # prost
 prost = { version = "0.5", optional = true }
@@ -62,18 +62,19 @@ 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 }
+hyper = { git = "https://github.com/hyperium/hyper", features = ["stream"], optional = true }
-tokio = { version = "=0.2.0-alpha.6", default-features = false, features = ["tcp"], optional = true }
+tokio = { version = "0.2", features = ["tcp"], optional = true }
-tower = { version = "=0.3.0-alpha.2", optional = true}
+tower = { git = "https://github.com/tower-rs/tower", optional = true}
-tower-make = "=0.3.0-alpha.2a"
+tower-make = { version = "0.3", features = ["connect"] }
-tower-balance =  { version = "=0.3.0-alpha.2", optional = true }
+tower-balance =  { git = "https://github.com/tower-rs/tower", optional = true }
-tower-load = { version = "=0.3.0-alpha.2", 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"
@@ -81,3 +82,5 @@ 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,7 +4,7 @@
 //! 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,
@@ -12,8 +12,6 @@ use std::{
     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>,

tonic/src/codec/decode.rs

@@ -1,6 +1,6 @@
 use super::Decoder;
 use crate::{body::BoxBody, metadata::MetadataMap, Code, Status};
-use bytes::{Buf, BufMut, Bytes, BytesMut, IntoBuf};
+use bytes::{Buf, BufMut, BytesMut};
 use futures_core::Stream;
 use futures_util::{future, ready};
 use http::StatusCode;
@@ -46,7 +46,7 @@ impl<T> Streaming<T> {
     pub(crate) fn new_response<B, D>(decoder: D, body: B, status_code: StatusCode) -> Self
     where
         B: Body + Send + Sync + 'static,
-        B::Data: Into<Bytes>,
+        // B::Data: Into<Bytes>,
         B::Error: Into<crate::Error>,
         D: Decoder<Item = T, Error = Status> + Send + Sync + 'static,
     {
@@ -56,7 +56,7 @@ impl<T> Streaming<T> {
     pub(crate) fn new_empty<B, D>(decoder: D, body: B) -> Self
     where
         B: Body + Send + Sync + 'static,
-        B::Data: Into<Bytes>,
+        // B::Data: Into<Bytes>,
         B::Error: Into<crate::Error>,
         D: Decoder<Item = T, Error = Status> + Send + Sync + 'static,
     {
@@ -66,7 +66,7 @@ impl<T> Streaming<T> {
     pub(crate) fn new_request<B, D>(decoder: D, body: B) -> Self
     where
         B: Body + Send + Sync + 'static,
-        B::Data: Into<Bytes>,
+        // B::Data: Into<Bytes>,
         B::Error: Into<crate::Error>,
         D: Decoder<Item = T, Error = Status> + Send + Sync + 'static,
     {
@@ -76,7 +76,7 @@ impl<T> Streaming<T> {
     fn new<B, D>(decoder: D, body: B, direction: Direction) -> Self
     where
         B: Body + Send + Sync + 'static,
-        B::Data: Into<Bytes>,
+        // B::Data: Into<Bytes>,
         B::Error: Into<crate::Error>,
         D: Decoder<Item = T, Error = Status> + Send + Sync + 'static,
     {
@@ -154,14 +154,12 @@ impl<T> Streaming<T> {
     }
 
     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 {
+            if self.buf.remaining() < 5 {
                 return Ok(None);
             }
 
-            let is_compressed = match buf.get_u8() {
+            let is_compressed = match self.buf.get_u8() {
                 0 => false,
                 1 => {
                     trace!("message compressed, compression not supported yet");
@@ -178,7 +176,7 @@ impl<T> Streaming<T> {
                     ));
                 }
             };
-            let len = buf.get_u32_be() as usize;
+            let len = self.buf.get_u32() as usize;
 
             self.state = State::ReadBody {
                 compression: is_compressed,
@@ -189,13 +187,10 @@ impl<T> Streaming<T> {
         if let State::ReadBody { len, .. } = &self.state {
             // if we haven't read enough of the message then return and keep
             // reading
-            if buf.remaining() < *len || self.buf.len() < *len + 5 {
+            if self.buf.remaining() < *len || self.buf.len() < *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;
@@ -251,8 +246,7 @@ impl<T> Stream for Streaming<T> {
                 self.buf.put(data);
             } else {
                 // FIXME: improve buf usage.
-                let buf1 = (&self.buf[..]).into_buf();
+                if self.buf.has_remaining() {
-                if buf1.has_remaining() {
                     trace!("unexpected EOF decoding stream");
                     Err(Status::new(
                         Code::Internal,

tonic/src/codec/encode.rs

@@ -1,5 +1,5 @@
-use crate::{body::BytesBuf, Code, Status};
+use crate::{Code, Status};
-use bytes::{BufMut, BytesMut, IntoBuf};
+use bytes::{BufMut, Bytes, BytesMut};
 use futures_core::{Stream, TryStream};
 use futures_util::{ready, StreamExt, TryStreamExt};
 use http::HeaderMap;
@@ -7,14 +7,14 @@ use http_body::Body;
 use pin_project::pin_project;
 use std::pin::Pin;
 use std::task::{Context, Poll};
-use tokio_codec::Encoder;
+use tokio_util::codec::Encoder;
 
 const BUFFER_SIZE: usize = 8 * 1024;
 
 pub(crate) fn encode_server<T, U>(
     encoder: T,
     source: U,
-) -> EncodeBody<impl Stream<Item = Result<BytesBuf, Status>>>
+) -> EncodeBody<impl Stream<Item = Result<Bytes, Status>>>
 where
     T: Encoder<Error = Status> + Send + Sync + 'static,
     T::Item: Send + Sync,
@@ -27,7 +27,7 @@ where
 pub(crate) fn encode_client<T, U>(
     encoder: T,
     source: U,
-) -> EncodeBody<impl Stream<Item = Result<BytesBuf, Status>>>
+) -> EncodeBody<impl Stream<Item = Result<Bytes, Status>>>
 where
     T: Encoder<Error = Status> + Send + Sync + 'static,
     T::Item: Send + Sync,
@@ -37,7 +37,7 @@ where
     EncodeBody::new_client(stream)
 }
 
-fn encode<T, U>(mut encoder: T, source: U) -> impl TryStream<Ok = BytesBuf, Error = Status>
+fn encode<T, U>(mut encoder: T, source: U) -> impl TryStream<Ok = Bytes, Error = Status>
 where
     T: Encoder<Error = Status>,
     U: Stream<Item = Result<T::Item, Status>>,
@@ -59,12 +59,12 @@ where
                     let len = buf.len() - 5;
                     assert!(len <= std::u32::MAX as usize);
                     {
-                        let mut cursor = std::io::Cursor::new(&mut buf[..5]);
+                        let mut buf = &mut buf[..5];
-                        cursor.put_u8(0); // byte must be 0, reserve doesn't auto-zero
+                        buf.put_u8(0); // byte must be 0, reserve doesn't auto-zero
-                        cursor.put_u32_be(len as u32);
+                        buf.put_u32(len as u32);
                     }
 
-                    yield Ok(buf.split_to(len + 5).freeze().into_buf());
+                    yield Ok(buf.split_to(len + 5).freeze());
                 },
                 Some(Err(status)) => yield Err(status),
                 None => break,
@@ -90,7 +90,7 @@ pub(crate) struct EncodeBody<S> {
 
 impl<S> EncodeBody<S>
 where
-    S: Stream<Item = Result<crate::body::BytesBuf, Status>> + Send + Sync + 'static,
+    S: Stream<Item = Result<Bytes, Status>> + Send + Sync + 'static,
 {
     pub(crate) fn new_client(inner: S) -> Self {
         Self {
@@ -111,9 +111,9 @@ where
 
 impl<S> Body for EncodeBody<S>
 where
-    S: Stream<Item = Result<crate::body::BytesBuf, Status>>,
+    S: Stream<Item = Result<Bytes, Status>>,
 {
-    type Data = BytesBuf;
+    type Data = Bytes;
     type Error = Status;
 
     fn is_end_stream(&self) -> bool {

tonic/src/codec/mod.rs

@@ -16,7 +16,7 @@ pub(crate) use self::encode::{encode_client, encode_server};
 #[cfg(feature = "prost")]
 #[cfg_attr(docsrs, doc(cfg(feature = "prost")))]
 pub use self::prost::ProstCodec;
-pub use tokio_codec::{Decoder, Encoder};
+pub use tokio_util::codec::{Decoder, Encoder};
 
 use crate::Status;
 

tonic/src/codec/prost.rs

@@ -1,6 +1,6 @@
 use super::{Codec, Decoder, Encoder};
 use crate::{Code, Status};
-use bytes::{BufMut, BytesMut};
+use bytes::{Buf, BufMut, BytesMut};
 use prost::Message;
 use std::marker::PhantomData;
 
@@ -51,8 +51,14 @@ impl<T: Message> Encoder for ProstEncoder<T> {
             buf.reserve(len);
         }
 
-        item.encode(buf)
+        let mut v = Vec::with_capacity(len);
-            .map_err(|_| unreachable!("Message only errors if not enough space"))
+
+        item.encode(&mut v)
+            .expect("Message only errors if not enough space");
+
+        buf.extend(v);
+
+        Ok(())
     }
 }
 
@@ -65,9 +71,19 @@ impl<U: Message + Default> Decoder for ProstDecoder<U> {
     type Error = Status;
 
     fn decode(&mut self, buf: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
-        Message::decode(buf.take())
+        let mut cursor = std::io::Cursor::new(&buf[..]);
+
+        let item = Message::decode(&mut cursor)
             .map(Option::Some)
-            .map_err(from_decode_error)
+            .map_err(from_decode_error)?;
+
+        let amt = cursor.position() as usize;
+
+        drop(cursor);
+
+        buf.advance(amt);
+
+        Ok(item)
     }
 }
 

tonic/src/codec/tests.rs

@@ -1,145 +1,140 @@
-use super::{
+// use super::{
-    encode_server,
+//     encode_server,
-    prost::{ProstDecoder, ProstEncoder},
+//     prost::{ProstDecoder, ProstEncoder},
-    Streaming,
+//     Streaming,
-};
+// };
-use crate::Status;
+// use crate::Status;
-use bytes::{Buf, BufMut, Bytes, BytesMut, IntoBuf};
+// use bytes04 as bytes;
-use http_body::Body;
+// use bytes04::{Buf, BufMut, Bytes, BytesMut};
-use prost::Message;
+// use http_body::Body;
-use std::{
+// use prost::Message;
-    io::Cursor,
+// use std::{
-    pin::Pin,
+//     io::Cursor,
-    task::{Context, Poll},
+//     pin::Pin,
-};
+//     task::{Context, Poll},
+// };
 
-#[derive(Clone, PartialEq, prost::Message)]
+// #[derive(Clone, PartialEq, prost::Message)]
-struct Msg {
+// struct Msg {
-    #[prost(bytes, tag = "1")]
+//     #[prost(bytes, tag = "1")]
-    data: Vec<u8>,
+//     data: Vec<u8>,
-}
+// }
 
-#[tokio::test]
+// #[tokio::test]
-async fn decode() {
+// async fn decode() {
-    let decoder = ProstDecoder::<Msg>::default();
+//     let decoder = ProstDecoder::<Msg>::default();
 
-    let data = vec![0u8; 10000];
+//     let data = vec![0u8; 10000];
-    let data_len = data.len();
+//     let data_len = data.len();
-    let msg = Msg { data };
+//     let msg = Msg { data };
 
-    let mut buf = BytesMut::new();
+//     let mut buf = BytesMut::new();
-    let len = msg.encoded_len();
+//     let len = msg.encoded_len();
 
-    buf.reserve(len + 5);
+//     buf.reserve(len + 5);
-    buf.put_u8(0);
+//     buf.put_u8(0);
-    buf.put_u32_be(len as u32);
+//     buf.put_u32_be(len as u32);
-    msg.encode(&mut buf).unwrap();
 
-    let body = MockBody {
+//     msg.encode(&mut buf).unwrap();
-        data: buf.freeze(),
-        partial_len: 10005,
-        count: 0,
-    };
 
-    let mut stream = Streaming::new_request(decoder, body);
+//     let body = body::MockBody::new(&buf[..], 10005, 0);
 
-    let mut i = 0usize;
+//     let mut stream = Streaming::new_request(decoder, body);
-    while let Some(msg) = stream.message().await.unwrap() {
-        assert_eq!(msg.data.len(), data_len);
-        i += 1;
-    }
-    assert_eq!(i, 1);
-}
 
-#[tokio::test]
+//     let mut i = 0usize;
-async fn encode() {
+//     while let Some(msg) = stream.message().await.unwrap() {
-    let encoder = ProstEncoder::<Msg>::default();
+//         assert_eq!(msg.data.len(), data_len);
+//         i += 1;
+//     }
+//     assert_eq!(i, 1);
+// }
 
-    let data = Vec::from(&[0u8; 1024][..]);
+// #[tokio::test]
-    let msg = Msg { data };
+// async fn encode() {
+//     let encoder = ProstEncoder::<Msg>::default();
 
-    let messages = std::iter::repeat(Ok::<_, Status>(msg)).take(10000);
+//     let data = Vec::from(&[0u8; 1024][..]);
-    let source = futures_util::stream::iter(messages);
+//     let msg = Msg { data };
 
-    let body = encode_server(encoder, source);
+//     let messages = std::iter::repeat(Ok::<_, Status>(msg)).take(10000);
+//     let source = futures_util::stream::iter(messages);
 
-    futures_util::pin_mut!(body);
+//     let body = encode_server(encoder, source);
 
-    while let Some(r) = body.next().await {
+//     futures_util::pin_mut!(body);
-        r.unwrap();
-    }
-}
 
-#[derive(Debug)]
+//     while let Some(r) = body.next().await {
-struct MockBody {
+//         r.unwrap();
-    data: Bytes,
+//     }
+// }
 
-    // the size of the partial message to send
+// mod body {
-    partial_len: usize,
+//     use crate::Status;
+//     use bytes::Bytes;
+//     use http_body::Body;
+//     use std::{
+//         pin::Pin,
+//         task::{Context, Poll},
+//     };
 
-    // the number of times we've sent
+//     #[derive(Debug)]
-    count: usize,
+//     pub struct MockBody {
-}
+//         data: Bytes,
 
-impl Body for MockBody {
+//         // the size of the partial message to send
-    type Data = Data;
+//         partial_len: usize,
-    type Error = Status;
 
-    fn poll_data(
+//         // the number of times we've sent
-        mut self: Pin<&mut Self>,
+//         count: usize,
-        cx: &mut Context<'_>,
+//     }
-    ) -> Poll<Option<Result<Self::Data, Self::Error>>> {
-        // every other call to poll_data returns data
-        let should_send = self.count % 2 == 0;
-        let data_len = self.data.len();
-        let partial_len = self.partial_len;
-        let count = self.count;
-        if data_len > 0 {
-            let result = if should_send {
-                let response = self
-                    .data
-                    .split_to(if count == 0 { partial_len } else { data_len })
-                    .into_buf();
-                Poll::Ready(Some(Ok(Data(response))))
-            } else {
-                cx.waker().wake_by_ref();
-                Poll::Pending
-            };
-            // make some fake progress
-            self.count += 1;
-            result
-        } else {
-            Poll::Ready(None)
-        }
-    }
 
-    fn poll_trailers(
+//     impl MockBody {
-        self: Pin<&mut Self>,
+//         pub fn new(b: &[u8], partial_len: usize, count) -> Self {
-        cx: &mut Context<'_>,
+//             MockBody {
-    ) -> Poll<Result<Option<http::HeaderMap>, Self::Error>> {
+//                 data: Bytes::copy_from_slice(&b[..]),
-        drop(cx);
+//                 partial_len,
-        Poll::Ready(Ok(None))
+//                 count
-    }
+//             }
-}
+//         }
+//     }
 
-struct Data(Cursor<Bytes>);
+//     impl Body for MockBody {
+//         type Data = Bytes;
+//         type Error = Status;
 
-impl Into<Bytes> for Data {
+//         fn poll_data(
-    fn into(self) -> Bytes {
+//             mut self: Pin<&mut Self>,
-        self.0.into_inner()
+//             cx: &mut Context<'_>,
-    }
+//         ) -> Poll<Option<Result<Self::Data, Self::Error>>> {
-}
+//             // every other call to poll_data returns data
+//             let should_send = self.count % 2 == 0;
+//             let data_len = self.data.len();
+//             let partial_len = self.partial_len;
+//             let count = self.count;
+//             if data_len > 0 {
+//                 let result = if should_send {
+//                     let response =
+//                         self.data
+//                             .split_to(if count == 0 { partial_len } else { data_len });
+//                     Poll::Ready(Some(Ok(response)))
+//                 } else {
+//                     cx.waker().wake_by_ref();
+//                     Poll::Pending
+//                 };
+//                 // make some fake progress
+//                 self.count += 1;
+//                 result
+//             } else {
+//                 Poll::Ready(None)
+//             }
+//         }
 
-impl Buf for Data {
+//         fn poll_trailers(
-    fn remaining(&self) -> usize {
+//             self: Pin<&mut Self>,
-        self.0.remaining()
+//             cx: &mut Context<'_>,
-    }
+//         ) -> Poll<Result<Option<http::HeaderMap>, Self::Error>> {
-
+//             drop(cx);
-    fn bytes(&self) -> &[u8] {
+//             Poll::Ready(Ok(None))
-        self.0.bytes()
+//         }
-    }
+//     }
-
+// }
-    fn advance(&mut self, cnt: usize) {
-        self.0.advance(cnt)
-    }
-}

tonic/src/metadata/encoding.rs

@@ -70,7 +70,7 @@ impl self::value_encoding::Sealed for Ascii {
     }
 
     fn from_shared(value: Bytes) -> Result<HeaderValue, InvalidMetadataValueBytes> {
-        HeaderValue::from_shared(value).map_err(|_| InvalidMetadataValueBytes::new())
+        HeaderValue::from_maybe_shared(value).map_err(|_| InvalidMetadataValueBytes::new())
     }
 
     fn from_static(value: &'static str) -> HeaderValue {
@@ -78,7 +78,7 @@ impl self::value_encoding::Sealed for Ascii {
     }
 
     fn decode(value: &[u8]) -> Result<Bytes, InvalidMetadataValueBytes> {
-        Ok(Bytes::from(value))
+        Ok(Bytes::copy_from_slice(value))
     }
 
     fn equals(a: &HeaderValue, b: &[u8]) -> bool {
@@ -112,7 +112,8 @@ impl self::value_encoding::Sealed for Binary {
 
     fn from_bytes(value: &[u8]) -> Result<HeaderValue, InvalidMetadataValueBytes> {
         let encoded_value: String = base64::encode_config(value, base64::STANDARD_NO_PAD);
-        HeaderValue::from_shared(encoded_value.into()).map_err(|_| InvalidMetadataValueBytes::new())
+        HeaderValue::from_maybe_shared(Bytes::from(encoded_value))
+            .map_err(|_| InvalidMetadataValueBytes::new())
     }
 
     fn from_shared(value: Bytes) -> Result<HeaderValue, InvalidMetadataValueBytes> {
@@ -126,7 +127,7 @@ impl self::value_encoding::Sealed for Binary {
         unsafe {
             // Because this is valid base64 this must be a valid HTTP header value,
             // no need to check again by calling from_shared.
-            HeaderValue::from_shared_unchecked(Bytes::from_static(value.as_ref()))
+            HeaderValue::from_maybe_shared_unchecked(Bytes::from_static(value.as_ref()))
         }
     }
 

tonic/src/metadata/key.rs

@@ -193,7 +193,7 @@ impl<'a, VE: ValueEncoding> From<&'a MetadataKey<VE>> for MetadataKey<VE> {
 impl<VE: ValueEncoding> From<MetadataKey<VE>> for Bytes {
     #[inline]
     fn from(name: MetadataKey<VE>) -> Bytes {
-        name.inner.into()
+        Bytes::copy_from_slice(name.inner.as_ref())
     }
 }
 

tonic/src/metadata/map.rs

@@ -2176,9 +2176,7 @@ mod as_metadata_key {
             self,
             map: &mut MetadataMap,
         ) -> Result<Entry<'_, HeaderValue>, InvalidMetadataKey> {
-            map.headers
+            Ok(map.headers.entry(self.inner))
-                .entry(self.inner)
-                .map_err(|_| InvalidMetadataKey::new())
         }
 
         #[doc(hidden)]
@@ -2221,9 +2219,7 @@ mod as_metadata_key {
             self,
             map: &mut MetadataMap,
         ) -> Result<Entry<'_, HeaderValue>, InvalidMetadataKey> {
-            map.headers
+            Ok(map.headers.entry(&self.inner))
-                .entry(&self.inner)
-                .map_err(|_| InvalidMetadataKey::new())
         }
 
         #[doc(hidden)]
@@ -2278,9 +2274,11 @@ mod as_metadata_key {
             if !VE::is_valid_key(self) {
                 return Err(InvalidMetadataKey::new());
             }
-            map.headers
+
-                .entry(self)
+            let key = http::header::HeaderName::from_bytes(self.as_bytes())
-                .map_err(|_| InvalidMetadataKey::new())
+                .map_err(|_| InvalidMetadataKey::new())?;
+            let entry = map.headers.entry(key);
+            Ok(entry)
         }
 
         #[doc(hidden)]
@@ -2338,9 +2336,10 @@ mod as_metadata_key {
             if !VE::is_valid_key(self.as_str()) {
                 return Err(InvalidMetadataKey::new());
             }
-            map.headers
+
-                .entry(self.as_str())
+            let key = http::header::HeaderName::from_bytes(self.as_bytes())
-                .map_err(|_| InvalidMetadataKey::new())
+                .map_err(|_| InvalidMetadataKey::new())?;
+            Ok(map.headers.entry(key))
         }
 
         #[doc(hidden)]
@@ -2398,9 +2397,10 @@ mod as_metadata_key {
             if !VE::is_valid_key(self) {
                 return Err(InvalidMetadataKey::new());
             }
-            map.headers
+
-                .entry(self.as_str())
+            let key = http::header::HeaderName::from_bytes(self.as_bytes())
-                .map_err(|_| InvalidMetadataKey::new())
+                .map_err(|_| InvalidMetadataKey::new())?;
+            Ok(map.headers.entry(key))
         }
 
         #[doc(hidden)]

tonic/src/metadata/value.rs

@@ -139,7 +139,7 @@ impl<VE: ValueEncoding> MetadataValue<VE> {
     #[inline]
     pub unsafe fn from_shared_unchecked(src: Bytes) -> Self {
         MetadataValue {
-            inner: HeaderValue::from_shared_unchecked(src),
+            inner: HeaderValue::from_maybe_shared_unchecked(src),
             phantom: PhantomData,
         }
     }
@@ -510,7 +510,7 @@ impl FromStr for MetadataValue<Ascii> {
 impl<VE: ValueEncoding> From<MetadataValue<VE>> for Bytes {
     #[inline]
     fn from(value: MetadataValue<VE>) -> Bytes {
-        Bytes::from(value.inner)
+        Bytes::copy_from_slice(value.inner.as_bytes())
     }
 }
 

tonic/src/status.rs

@@ -325,7 +325,7 @@ impl Status {
                 .unwrap_or_else(|| Ok(String::new()));
             let details = header_map
                 .get(GRPC_STATUS_DETAILS_HEADER)
-                .map(|h| Bytes::from(h.as_bytes()))
+                .map(|h| Bytes::copy_from_slice(h.as_bytes()))
                 .unwrap_or_else(Bytes::new);
             match error_message {
                 Ok(message) => Status {
@@ -380,19 +380,19 @@ impl Status {
                     .to_string()
                     .into()
             } else {
-                Bytes::from(self.message().as_bytes())
+                Bytes::copy_from_slice(self.message().as_bytes())
             };
 
             header_map.insert(
                 GRPC_STATUS_MESSAGE_HEADER,
-                HeaderValue::from_shared(to_write).map_err(invalid_header_value_byte)?,
+                HeaderValue::from_maybe_shared(to_write).map_err(invalid_header_value_byte)?,
             );
         }
 
         if !self.details.is_empty() {
             header_map.insert(
                 GRPC_STATUS_DETAILS_HEADER,
-                HeaderValue::from_shared(self.details.clone())
+                HeaderValue::from_maybe_shared(self.details.clone())
                     .map_err(invalid_header_value_byte)?,
             );
         }

tonic/src/transport/channel.rs

@@ -7,7 +7,7 @@ use super::{
 use crate::{body::BoxBody, client::GrpcService};
 use bytes::Bytes;
 use http::{
-    uri::{InvalidUriBytes, Uri},
+    uri::{InvalidUri, Uri},
     Request, Response,
 };
 use std::{
@@ -88,8 +88,8 @@ impl Channel {
     /// # use tonic::transport::Channel;
     /// Channel::from_shared("https://example.com");
     /// ```
-    pub fn from_shared(s: impl Into<Bytes>) -> Result<Endpoint, InvalidUriBytes> {
+    pub fn from_shared(s: impl Into<Bytes>) -> Result<Endpoint, InvalidUri> {
-        let uri = Uri::from_shared(s.into())?;
+        let uri = Uri::from_maybe_shared(s.into())?;
         Ok(Self::builder(uri))
     }
 

tonic/src/transport/endpoint.rs

@@ -5,7 +5,7 @@ use super::{
     tls::{Certificate, Identity},
 };
 use bytes::Bytes;
-use http::uri::{InvalidUriBytes, Uri};
+use http::uri::{InvalidUri, Uri};
 use std::{
     convert::{TryFrom, TryInto},
     fmt,
@@ -63,8 +63,8 @@ impl Endpoint {
     /// # use tonic::transport::Endpoint;
     /// Endpoint::from_shared("https://example.com".to_string());
     /// ```
-    pub fn from_shared(s: impl Into<Bytes>) -> Result<Self, InvalidUriBytes> {
+    pub fn from_shared(s: impl Into<Bytes>) -> Result<Self, InvalidUri> {
-        let uri = Uri::from_shared(s.into())?;
+        let uri = Uri::from_maybe_shared(s.into())?;
         Ok(Self::from(uri))
     }
 
@@ -179,7 +179,7 @@ impl From<Uri> for Endpoint {
 }
 
 impl TryFrom<Bytes> for Endpoint {
-    type Error = InvalidUriBytes;
+    type Error = InvalidUri;
 
     fn try_from(t: Bytes) -> Result<Self, Self::Error> {
         Self::from_shared(t)
@@ -187,7 +187,7 @@ impl TryFrom<Bytes> for Endpoint {
 }
 
 impl TryFrom<String> for Endpoint {
-    type Error = InvalidUriBytes;
+    type Error = InvalidUri;
 
     fn try_from(t: String) -> Result<Self, Self::Error> {
         Self::from_shared(t.into_bytes())

tonic/src/transport/server.rs

@@ -5,7 +5,10 @@ use super::service::{layer_fn, BoxedIo, Or, Routes, ServiceBuilderExt};
 use super::{service::TlsAcceptor, tls::Identity, Certificate};
 use crate::body::BoxBody;
 use futures_core::Stream;
-use futures_util::{future, ready, try_future::MapErr, TryFutureExt, TryStreamExt};
+use futures_util::{
+    future::{self, MapErr},
+    ready, TryFutureExt, TryStreamExt,
+};
 use http::{Request, Response};
 use hyper::{
     server::{accept::Accept, conn},
@@ -21,7 +24,7 @@ use std::{
     // time::Duration,
 };
 use tower::{
-    layer::{util::Stack, Layer},
+    layer::{Layer, Stack},
     limit::concurrency::ConcurrencyLimitLayer,
     // timeout::TimeoutLayer,
     Service,
@@ -203,28 +206,30 @@ impl Server {
         let max_concurrent_streams = self.max_concurrent_streams;
         // let timeout = self.timeout.clone();
 
-        let incoming = hyper::server::accept::from_stream(async_stream::try_stream! {
+        let incoming = hyper::server::accept::from_stream::<_, _, crate::Error>(
-            let mut tcp = TcpIncoming::bind(addr)?;
+            async_stream::try_stream! {
+                let mut tcp = TcpIncoming::bind(addr)?;
 
-            while let Some(stream) = tcp.try_next().await? {
+                while let Some(stream) = tcp.try_next().await? {
-                #[cfg(feature = "tls")]
+                    #[cfg(feature = "tls")]
-                {
+                    {
-                    if let Some(tls) = &self.tls {
+                        if let Some(tls) = &self.tls {
-                        let io = match tls.connect(stream.into_inner()).await {
+                            let io = match tls.connect(stream.into_inner()).await {
-                            Ok(io) => io,
+                                Ok(io) => io,
-                            Err(error) => {
+                                Err(error) => {
-                                error!(message = "Unable to accept incoming connection.", %error);
+                                    error!(message = "Unable to accept incoming connection.", %error);
-                                continue
+                                    continue
-                            },
+                                },
-                        };
+                            };
-                        yield BoxedIo::new(io);
+                            yield BoxedIo::new(io);
-                        continue;
+                            continue;
+                        }
                     }
-                }
 
-                yield BoxedIo::new(stream);
+                    yield BoxedIo::new(stream);
-            }
+                }
-        });
+            },
+        );
 
         let svc = MakeSvc {
             inner: svc,

tonic/src/transport/service/either.rs

@@ -1,4 +1,4 @@
-use futures_util::try_future::{MapErr, TryFutureExt};
+use futures_util::future::{MapErr, TryFutureExt};
 use std::{
     future::Future,
     pin::Pin,

tonic/src/transport/service/layer.rs

@@ -1,6 +1,6 @@
 use super::either::Either;
 use tower::{
-    layer::{util::Stack, Layer},
+    layer::{Layer, Stack},
     ServiceBuilder,
 };
 pub(crate) trait ServiceBuilderExt<L> {

tonic/src/transport/service/router.rs

@@ -1,6 +1,6 @@
 use futures_util::{
     future::Either,
-    try_future::{MapErr, TryFutureExt},
+    future::{MapErr, TryFutureExt},
 };
 use std::{
     fmt,