chore: More clean up (#172)

* Clean up client and codec

* Clean up codegen, server and upgrade hyper

tonic/Cargo.toml

@@ -62,7 +62,7 @@ prost-derive = { version = "0.5", optional = true }
 async-trait = { version = "0.1.13", optional = true }
 
 # transport
-hyper = { git = "https://github.com/hyperium/hyper", features = ["stream"], optional = true }
+hyper = { version = "0.13", 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"] }

tonic/src/body.rs

@@ -101,7 +101,6 @@ impl BoxBody {
     pub fn map_from<B>(inner: B) -> Self
     where
         B: Body + Send + Sync + 'static,
-        // B::Data: Into<Bytes>,
         B::Error: Into<crate::Error>,
     {
         BoxBody {
@@ -143,7 +142,6 @@ impl HttpBody for BoxBody {
 impl<B> HttpBody for MapBody<B>
 where
     B: Body,
-    // B::Data: Into<Bytes>,
     B::Error: Into<crate::Error>,
 {
     type Data = Bytes;
@@ -206,17 +204,15 @@ impl HttpBody for EmptyBody {
 
     fn poll_data(
         self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
+        _cx: &mut Context<'_>,
     ) -> Poll<Option<Result<Self::Data, Self::Error>>> {
-        drop(cx);
         Poll::Ready(None)
     }
 
     fn poll_trailers(
         self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
+        _cx: &mut Context<'_>,
     ) -> Poll<Result<Option<http::HeaderMap>, Self::Error>> {
-        drop(cx);
         Poll::Ready(Ok(None))
     }
 }

tonic/src/client/grpc.rs

@@ -59,7 +59,6 @@ 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>,
         C: Codec<Encode = M1, Decode = M2>,
         M1: Send + Sync + 'static,
         M2: Send + Sync + 'static,
@@ -79,7 +78,6 @@ 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>,
         S: Stream<Item = M1> + Send + Sync + 'static,
         C: Codec<Encode = M1, Decode = M2>,
         M1: Send + Sync + 'static,
@@ -112,7 +110,6 @@ 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>,
         C: Codec<Encode = M1, Decode = M2>,
         M1: Send + Sync + 'static,
         M2: Send + Sync + 'static,
@@ -131,7 +128,6 @@ 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>::Error: Into<crate::Error>,
         S: Stream<Item = M1> + Send + Sync + 'static,
         C: Codec<Encode = M1, Decode = M2>,
@@ -200,8 +196,8 @@ impl<T: Clone> Clone for Grpc<T> {
     }
 }
 
-impl<T> fmt::Debug for Grpc<T> {
+impl<T: fmt::Debug> fmt::Debug for Grpc<T> {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        f.debug_struct("Grpc").finish()
+        f.debug_struct("Grpc").field("inner", &self.inner).finish()
     }
 }

tonic/src/codec/decode.rs

@@ -46,7 +46,6 @@ 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::Error: Into<crate::Error>,
         D: Decoder<Item = T, Error = Status> + Send + Sync + 'static,
     {
@@ -56,7 +55,6 @@ 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::Error: Into<crate::Error>,
         D: Decoder<Item = T, Error = Status> + Send + Sync + 'static,
     {
@@ -66,7 +64,6 @@ 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::Error: Into<crate::Error>,
         D: Decoder<Item = T, Error = Status> + Send + Sync + 'static,
     {
@@ -76,7 +73,6 @@ 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::Error: Into<crate::Error>,
         D: Decoder<Item = T, Error = Status> + Send + Sync + 'static,
     {

tonic/src/codec/tests.rs

@@ -1,140 +1,148 @@
-// use super::{
+use super::{encode_server, Decoder, Encoder, Streaming};
-//     encode_server,
+use crate::Status;
-//     prost::{ProstDecoder, ProstEncoder},
+use bytes::{Buf, BufMut, BytesMut};
-//     Streaming,
+use http_body::Body;
-// };
-// use crate::Status;
-// use bytes04 as bytes;
-// use bytes04::{Buf, BufMut, Bytes, BytesMut};
-// use http_body::Body;
-// use prost::Message;
-// use std::{
-//     io::Cursor,
-//     pin::Pin,
-//     task::{Context, Poll},
-// };
 
-// #[derive(Clone, PartialEq, prost::Message)]
+const LEN: usize = 10000;
-// struct Msg {
-//     #[prost(bytes, tag = "1")]
-//     data: Vec<u8>,
-// }
 
-// #[tokio::test]
+#[tokio::test]
-// async fn decode() {
+async fn decode() {
-//     let decoder = ProstDecoder::<Msg>::default();
+    let decoder = MockDecoder::default();
 
-//     let data = vec![0u8; 10000];
+    let msg = vec![0u8; LEN];
-//     let data_len = data.len();
-//     let msg = Msg { data };
 
-//     let mut buf = BytesMut::new();
+    let mut buf = BytesMut::new();
-//     let len = msg.encoded_len();
 
-//     buf.reserve(len + 5);
+    buf.reserve(msg.len() + 5);
-//     buf.put_u8(0);
+    buf.put_u8(0);
-//     buf.put_u32_be(len as u32);
+    buf.put_u32(msg.len() as u32);
 
-//     msg.encode(&mut buf).unwrap();
+    buf.put(&msg[..]);
 
-//     let body = body::MockBody::new(&buf[..], 10005, 0);
+    let body = body::MockBody::new(&buf[..], 10005, 0);
 
-//     let mut stream = Streaming::new_request(decoder, body);
+    let mut stream = Streaming::new_request(decoder, body);
 
-//     let mut i = 0usize;
+    let mut i = 0usize;
-//     while let Some(msg) = stream.message().await.unwrap() {
+    while let Some(output_msg) = stream.message().await.unwrap() {
-//         assert_eq!(msg.data.len(), data_len);
+        assert_eq!(output_msg.len(), msg.len());
-//         i += 1;
+        i += 1;
-//     }
+    }
-//     assert_eq!(i, 1);
+    assert_eq!(i, 1);
-// }
+}
 
-// #[tokio::test]
+#[tokio::test]
-// async fn encode() {
+async fn encode() {
-//     let encoder = ProstEncoder::<Msg>::default();
+    let encoder = MockEncoder::default();
 
-//     let data = Vec::from(&[0u8; 1024][..]);
+    let msg = Vec::from(&[0u8; 1024][..]);
-//     let msg = Msg { data };
 
-//     let messages = std::iter::repeat(Ok::<_, Status>(msg)).take(10000);
+    let messages = std::iter::repeat(Ok::<_, Status>(msg)).take(10000);
-//     let source = futures_util::stream::iter(messages);
+    let source = futures_util::stream::iter(messages);
 
-//     let body = encode_server(encoder, source);
+    let body = encode_server(encoder, source);
 
-//     futures_util::pin_mut!(body);
+    futures_util::pin_mut!(body);
 
-//     while let Some(r) = body.next().await {
+    while let Some(r) = body.data().await {
-//         r.unwrap();
+        r.unwrap();
-//     }
+    }
-// }
+}
 
-// mod body {
+#[derive(Debug, Clone, Default)]
-//     use crate::Status;
+struct MockEncoder;
-//     use bytes::Bytes;
-//     use http_body::Body;
-//     use std::{
-//         pin::Pin,
-//         task::{Context, Poll},
-//     };
 
-//     #[derive(Debug)]
+impl Encoder for MockEncoder {
-//     pub struct MockBody {
+    type Item = Vec<u8>;
-//         data: Bytes,
+    type Error = Status;
 
-//         // the size of the partial message to send
+    fn encode(&mut self, item: Self::Item, buf: &mut BytesMut) -> Result<(), Self::Error> {
-//         partial_len: usize,
+        buf.put(&item[..]);
+        Ok(())
+    }
+}
 
-//         // the number of times we've sent
+#[derive(Debug, Clone, Default)]
-//         count: usize,
+struct MockDecoder;
-//     }
 
-//     impl MockBody {
+impl Decoder for MockDecoder {
-//         pub fn new(b: &[u8], partial_len: usize, count) -> Self {
+    type Item = Vec<u8>;
-//             MockBody {
+    type Error = Status;
-//                 data: Bytes::copy_from_slice(&b[..]),
-//                 partial_len,
-//                 count
-//             }
-//         }
-//     }
 
-//     impl Body for MockBody {
+    fn decode(&mut self, buf: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
-//         type Data = Bytes;
+        let out = Vec::from(&buf[..LEN]);
-//         type Error = Status;
+        buf.advance(LEN);
+        Ok(Some(out))
+    }
+}
 
-//         fn poll_data(
+mod body {
-//             mut self: Pin<&mut Self>,
+    use crate::Status;
-//             cx: &mut Context<'_>,
+    use bytes::Bytes;
-//         ) -> Poll<Option<Result<Self::Data, Self::Error>>> {
+    use http_body::Body;
-//             // every other call to poll_data returns data
+    use std::{
-//             let should_send = self.count % 2 == 0;
+        pin::Pin,
-//             let data_len = self.data.len();
+        task::{Context, Poll},
-//             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)
-//             }
-//         }
 
-//         fn poll_trailers(
+    #[derive(Debug)]
-//             self: Pin<&mut Self>,
+    pub(super) struct MockBody {
-//             cx: &mut Context<'_>,
+        data: Bytes,
-//         ) -> Poll<Result<Option<http::HeaderMap>, Self::Error>> {
+
-//             drop(cx);
+        // the size of the partial message to send
-//             Poll::Ready(Ok(None))
+        partial_len: usize,
-//         }
+
-//     }
+        // the number of times we've sent
-// }
+        count: usize,
+    }
+
+    impl MockBody {
+        pub(super) fn new(b: &[u8], partial_len: usize, count: usize) -> Self {
+            MockBody {
+                data: Bytes::copy_from_slice(&b[..]),
+                partial_len,
+                count,
+            }
+        }
+    }
+
+    impl Body for MockBody {
+        type Data = Bytes;
+        type Error = Status;
+
+        fn poll_data(
+            mut self: Pin<&mut Self>,
+            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)
+            }
+        }
+
+        fn poll_trailers(
+            self: Pin<&mut Self>,
+            cx: &mut Context<'_>,
+        ) -> Poll<Result<Option<http::HeaderMap>, Self::Error>> {
+            drop(cx);
+            Poll::Ready(Ok(None))
+        }
+    }
+}

tonic/src/codegen.rs

@@ -34,7 +34,3 @@ impl std::fmt::Display for Never {
 }
 
 impl std::error::Error for Never {}
-
-pub use bytes::*;
-pub use prost::*;
-pub use prost_derive::*;

tonic/src/server/grpc.rs

@@ -4,7 +4,6 @@ use crate::{
     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;
@@ -42,7 +41,6 @@ where
     where
         S: UnaryService<T::Decode, Response = T::Encode>,
         B: Body + Send + Sync + 'static,
-        B::Data: Into<Bytes> + Send,
         B::Error: Into<crate::Error> + Send,
     {
         let request = match self.map_request_unary(req).await {
@@ -73,7 +71,6 @@ where
         S: ServerStreamingService<T::Decode, Response = T::Encode>,
         S::ResponseStream: Send + Sync + 'static,
         B: Body + Send + Sync + 'static,
-        B::Data: Into<Bytes> + Send,
         B::Error: Into<crate::Error> + Send,
     {
         let request = match self.map_request_unary(req).await {
@@ -97,7 +94,6 @@ where
     where
         S: ClientStreamingService<T::Decode, Response = T::Encode>,
         B: Body + Send + Sync + 'static,
-        B::Data: Into<Bytes> + Send + 'static,
         B::Error: Into<crate::Error> + Send + 'static,
     {
         let request = self.map_request_streaming(req);
@@ -118,7 +114,6 @@ where
         S: StreamingService<T::Decode, Response = T::Encode> + Send,
         S::ResponseStream: Send + Sync + 'static,
         B: Body + Send + Sync + 'static,
-        B::Data: Into<Bytes> + Send,
         B::Error: Into<crate::Error> + Send,
     {
         let request = self.map_request_streaming(req);
@@ -132,7 +127,6 @@ where
     ) -> Result<Request<T::Decode>, Status>
     where
         B: Body + Send + Sync + 'static,
-        B::Data: Into<Bytes> + Send,
         B::Error: Into<crate::Error> + Send,
     {
         let (parts, body) = request.into_parts();
@@ -160,7 +154,6 @@ where
     ) -> Request<Streaming<T::Decode>>
     where
         B: Body + Send + Sync + 'static,
-        B::Data: Into<Bytes> + Send,
         B::Error: Into<crate::Error> + Send,
     {
         Request::from_http(request.map(|body| Streaming::new_request(self.codec.decoder(), body)))
@@ -203,8 +196,8 @@ where
     }
 }
 
-impl<T> fmt::Debug for Grpc<T> {
+impl<T: fmt::Debug> fmt::Debug for Grpc<T> {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        f.debug_struct("Grpc").finish()
+        f.debug_struct("Grpc").field("codec", &self.codec).finish()
     }
 }