From cce550be8530fdde39eeddded3b4b1350f1e3836 Mon Sep 17 00:00:00 2001 From: Lucio Franco Date: Wed, 11 Dec 2019 16:05:30 -0500 Subject: [PATCH] chore: More clean up (#172) * Clean up client and codec * Clean up codegen, server and upgrade hyper --- tonic/Cargo.toml | 2 +- tonic/src/body.rs | 8 +- tonic/src/client/grpc.rs | 8 +- tonic/src/codec/decode.rs | 4 - tonic/src/codec/tests.rs | 242 ++++++++++++++++++++------------------ tonic/src/codegen.rs | 4 - tonic/src/server/grpc.rs | 11 +- 7 files changed, 132 insertions(+), 147 deletions(-) diff --git a/tonic/Cargo.toml b/tonic/Cargo.toml index 11db22a..8f4e35e 100644 --- a/tonic/Cargo.toml +++ b/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"] } diff --git a/tonic/src/body.rs b/tonic/src/body.rs index d943935..1714cfb 100644 --- a/tonic/src/body.rs +++ b/tonic/src/body.rs @@ -101,7 +101,6 @@ impl BoxBody { pub fn map_from(inner: B) -> Self where B: Body + Send + Sync + 'static, - // B::Data: Into, B::Error: Into, { BoxBody { @@ -143,7 +142,6 @@ impl HttpBody for BoxBody { impl HttpBody for MapBody where B: Body, - // B::Data: Into, B::Error: Into, { 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>> { - drop(cx); Poll::Ready(None) } fn poll_trailers( self: Pin<&mut Self>, - cx: &mut Context<'_>, + _cx: &mut Context<'_>, ) -> Poll, Self::Error>> { - drop(cx); Poll::Ready(Ok(None)) } } diff --git a/tonic/src/client/grpc.rs b/tonic/src/client/grpc.rs index 45aee8d..07d90b8 100644 --- a/tonic/src/client/grpc.rs +++ b/tonic/src/client/grpc.rs @@ -59,7 +59,6 @@ impl Grpc { T: GrpcService, T::ResponseBody: Body + HttpBody + Send + 'static, ::Error: Into, - // ::Data: Into, C: Codec, M1: Send + Sync + 'static, M2: Send + Sync + 'static, @@ -79,7 +78,6 @@ impl Grpc { T: GrpcService, T::ResponseBody: Body + HttpBody + Send + 'static, ::Error: Into, - // ::Data: Into, S: Stream + Send + Sync + 'static, C: Codec, M1: Send + Sync + 'static, @@ -112,7 +110,6 @@ impl Grpc { T: GrpcService, T::ResponseBody: Body + HttpBody + Send + 'static, ::Error: Into, - // ::Data: Into, C: Codec, M1: Send + Sync + 'static, M2: Send + Sync + 'static, @@ -131,7 +128,6 @@ impl Grpc { where T: GrpcService, T::ResponseBody: Body + HttpBody + Send + 'static, - // ::Data: Into, ::Error: Into, S: Stream + Send + Sync + 'static, C: Codec, @@ -200,8 +196,8 @@ impl Clone for Grpc { } } -impl fmt::Debug for Grpc { +impl fmt::Debug for Grpc { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - f.debug_struct("Grpc").finish() + f.debug_struct("Grpc").field("inner", &self.inner).finish() } } diff --git a/tonic/src/codec/decode.rs b/tonic/src/codec/decode.rs index ed64c41..750a0dc 100644 --- a/tonic/src/codec/decode.rs +++ b/tonic/src/codec/decode.rs @@ -46,7 +46,6 @@ impl Streaming { pub(crate) fn new_response(decoder: D, body: B, status_code: StatusCode) -> Self where B: Body + Send + Sync + 'static, - // B::Data: Into, B::Error: Into, D: Decoder + Send + Sync + 'static, { @@ -56,7 +55,6 @@ impl Streaming { pub(crate) fn new_empty(decoder: D, body: B) -> Self where B: Body + Send + Sync + 'static, - // B::Data: Into, B::Error: Into, D: Decoder + Send + Sync + 'static, { @@ -66,7 +64,6 @@ impl Streaming { pub(crate) fn new_request(decoder: D, body: B) -> Self where B: Body + Send + Sync + 'static, - // B::Data: Into, B::Error: Into, D: Decoder + Send + Sync + 'static, { @@ -76,7 +73,6 @@ impl Streaming { fn new(decoder: D, body: B, direction: Direction) -> Self where B: Body + Send + Sync + 'static, - // B::Data: Into, B::Error: Into, D: Decoder + Send + Sync + 'static, { diff --git a/tonic/src/codec/tests.rs b/tonic/src/codec/tests.rs index a5292fe..80525d8 100644 --- a/tonic/src/codec/tests.rs +++ b/tonic/src/codec/tests.rs @@ -1,140 +1,148 @@ -// use super::{ -// encode_server, -// prost::{ProstDecoder, ProstEncoder}, -// Streaming, -// }; -// 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}, -// }; +use super::{encode_server, Decoder, Encoder, Streaming}; +use crate::Status; +use bytes::{Buf, BufMut, BytesMut}; +use http_body::Body; -// #[derive(Clone, PartialEq, prost::Message)] -// struct Msg { -// #[prost(bytes, tag = "1")] -// data: Vec, -// } +const LEN: usize = 10000; -// #[tokio::test] -// async fn decode() { -// let decoder = ProstDecoder::::default(); +#[tokio::test] +async fn decode() { + let decoder = MockDecoder::default(); -// let data = vec![0u8; 10000]; -// let data_len = data.len(); -// let msg = Msg { data }; + let msg = vec![0u8; LEN]; -// let mut buf = BytesMut::new(); -// let len = msg.encoded_len(); + let mut buf = BytesMut::new(); -// buf.reserve(len + 5); -// buf.put_u8(0); -// buf.put_u32_be(len as u32); + buf.reserve(msg.len() + 5); + buf.put_u8(0); + 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; -// while let Some(msg) = stream.message().await.unwrap() { -// assert_eq!(msg.data.len(), data_len); -// i += 1; -// } -// assert_eq!(i, 1); -// } + let mut i = 0usize; + while let Some(output_msg) = stream.message().await.unwrap() { + assert_eq!(output_msg.len(), msg.len()); + i += 1; + } + assert_eq!(i, 1); +} -// #[tokio::test] -// async fn encode() { -// let encoder = ProstEncoder::::default(); +#[tokio::test] +async fn encode() { + let encoder = MockEncoder::default(); -// let data = Vec::from(&[0u8; 1024][..]); -// let msg = Msg { data }; + let msg = Vec::from(&[0u8; 1024][..]); -// let messages = std::iter::repeat(Ok::<_, Status>(msg)).take(10000); -// let source = futures_util::stream::iter(messages); + let messages = std::iter::repeat(Ok::<_, Status>(msg)).take(10000); + 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 { -// r.unwrap(); -// } -// } + while let Some(r) = body.data().await { + r.unwrap(); + } +} -// mod body { -// use crate::Status; -// use bytes::Bytes; -// use http_body::Body; -// use std::{ -// pin::Pin, -// task::{Context, Poll}, -// }; +#[derive(Debug, Clone, Default)] +struct MockEncoder; -// #[derive(Debug)] -// pub struct MockBody { -// data: Bytes, +impl Encoder for MockEncoder { + type Item = Vec; + type Error = Status; -// // the size of the partial message to send -// partial_len: usize, + fn encode(&mut self, item: Self::Item, buf: &mut BytesMut) -> Result<(), Self::Error> { + buf.put(&item[..]); + Ok(()) + } +} -// // the number of times we've sent -// count: usize, -// } +#[derive(Debug, Clone, Default)] +struct MockDecoder; -// impl MockBody { -// pub fn new(b: &[u8], partial_len: usize, count) -> Self { -// MockBody { -// data: Bytes::copy_from_slice(&b[..]), -// partial_len, -// count -// } -// } -// } +impl Decoder for MockDecoder { + type Item = Vec; + type Error = Status; -// impl Body for MockBody { -// type Data = Bytes; -// type Error = Status; + fn decode(&mut self, buf: &mut BytesMut) -> Result, Self::Error> { + let out = Vec::from(&buf[..LEN]); + buf.advance(LEN); + Ok(Some(out)) + } +} -// fn poll_data( -// mut self: Pin<&mut Self>, -// cx: &mut Context<'_>, -// ) -> Poll>> { -// // 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) -// } -// } +mod body { + use crate::Status; + use bytes::Bytes; + use http_body::Body; + use std::{ + pin::Pin, + task::{Context, Poll}, + }; -// fn poll_trailers( -// self: Pin<&mut Self>, -// cx: &mut Context<'_>, -// ) -> Poll, Self::Error>> { -// drop(cx); -// Poll::Ready(Ok(None)) -// } -// } -// } + #[derive(Debug)] + pub(super) struct MockBody { + data: Bytes, + + // the size of the partial message to send + 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>> { + // 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, Self::Error>> { + drop(cx); + Poll::Ready(Ok(None)) + } + } +} diff --git a/tonic/src/codegen.rs b/tonic/src/codegen.rs index 52116d7..e5eea9b 100644 --- a/tonic/src/codegen.rs +++ b/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::*; diff --git a/tonic/src/server/grpc.rs b/tonic/src/server/grpc.rs index fc2f7bc..b099b36 100644 --- a/tonic/src/server/grpc.rs +++ b/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, B: Body + Send + Sync + 'static, - B::Data: Into + Send, B::Error: Into + Send, { let request = match self.map_request_unary(req).await { @@ -73,7 +71,6 @@ where S: ServerStreamingService, S::ResponseStream: Send + Sync + 'static, B: Body + Send + Sync + 'static, - B::Data: Into + Send, B::Error: Into + Send, { let request = match self.map_request_unary(req).await { @@ -97,7 +94,6 @@ where where S: ClientStreamingService, B: Body + Send + Sync + 'static, - B::Data: Into + Send + 'static, B::Error: Into + Send + 'static, { let request = self.map_request_streaming(req); @@ -118,7 +114,6 @@ where S: StreamingService + Send, S::ResponseStream: Send + Sync + 'static, B: Body + Send + Sync + 'static, - B::Data: Into + Send, B::Error: Into + Send, { let request = self.map_request_streaming(req); @@ -132,7 +127,6 @@ where ) -> Result, Status> where B: Body + Send + Sync + 'static, - B::Data: Into + Send, B::Error: Into + Send, { let (parts, body) = request.into_parts(); @@ -160,7 +154,6 @@ where ) -> Request> where B: Body + Send + Sync + 'static, - B::Data: Into + Send, B::Error: Into + Send, { Request::from_http(request.map(|body| Streaming::new_request(self.codec.decoder(), body))) @@ -203,8 +196,8 @@ where } } -impl fmt::Debug for Grpc { +impl fmt::Debug for Grpc { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - f.debug_struct("Grpc").finish() + f.debug_struct("Grpc").field("codec", &self.codec).finish() } }