diff --git a/tonic/src/body.rs b/tonic/src/body.rs index 04cf028..b9be632 100644 --- a/tonic/src/body.rs +++ b/tonic/src/body.rs @@ -96,7 +96,6 @@ impl HttpBody for BoxBody { type Error = Status; fn is_end_stream(&self) -> bool { - // Body::is_end_stream(&self.inner) self.inner.is_end_stream() } diff --git a/tonic/src/client/grpc.rs b/tonic/src/client/grpc.rs index 5b5cfec..80d164a 100644 --- a/tonic/src/client/grpc.rs +++ b/tonic/src/client/grpc.rs @@ -62,10 +62,8 @@ impl Grpc { ::Error: Into, ::Data: Into, C: Codec, - C::Encoder: Send + 'static, - C::Decoder: Send + 'static, M1: Send + 'static, - { + M2: Send + 'static { let request = request.map(|m| stream::once(future::ok(m))); self.client_streaming(request, path, codec).await } @@ -84,9 +82,8 @@ impl Grpc { ::Data: Into, S: Stream> + Send + 'static, C: Codec, - C::Encoder: Send + 'static, - C::Decoder: Send + 'static, - M1: Send, + M1: Send + 'static, + M2: Send + 'static { let (mut parts, body) = self.streaming(request, path, codec).await?.into_parts(); @@ -117,9 +114,8 @@ impl Grpc { ::Error: Into, ::Data: Into, C: Codec, - C::Encoder: Send + 'static, - C::Decoder: Send + 'static, M1: Send + 'static, + M2: Send + 'static { let request = request.map(|m| stream::once(future::ok(m))); self.streaming(request, path, codec).await @@ -139,9 +135,8 @@ impl Grpc { ::Error: Into, S: Stream> + Send + 'static, C: Codec, - C::Encoder: Send + 'static, - C::Decoder: Send + 'static, - M1: Send, + M1: Send + 'static, + M2: Send + 'static { let mut parts = Parts::default(); parts.path_and_query = Some(path); diff --git a/tonic/src/client/mod.rs b/tonic/src/client/mod.rs index 2a59b7e..091608e 100644 --- a/tonic/src/client/mod.rs +++ b/tonic/src/client/mod.rs @@ -1,4 +1,4 @@ -//! gRPC over HTTP2 client implementation. +//! gRPC client implementation. //! //! This module contains the low level components to build a gRPC client. It //! provides a codec agnostic gRPC client dispatcher and a decorated tower diff --git a/tonic/src/codec/decode.rs b/tonic/src/codec/decode.rs index b092dd5..a29715d 100644 --- a/tonic/src/codec/decode.rs +++ b/tonic/src/codec/decode.rs @@ -1,18 +1,17 @@ use super::Decoder; -use crate::body::BoxBody; -use crate::metadata::MetadataMap; -use crate::{Code, Status}; +use crate::{Code, Status, BoxBody, metadata::MetadataMap}; use bytes::{Buf, BufMut, Bytes, BytesMut, IntoBuf}; use futures_core::Stream; use futures_util::{future, ready}; use http::StatusCode; use http_body::Body; -use std::fmt; -use std::pin::Pin; -use std::task::{Context, Poll}; +use std::{fmt, pin::Pin, task::{Context, Poll}}; use tracing::{debug, trace}; -// #[derive(Debug)] +/// Streaming requests and responses. +/// +/// This will wrap some inner [`Body`] and [`Decoder`] and provide an interface +/// to fetch the message stream and trailing metadata pub struct Streaming { decoder: Box + Send + 'static>, body: BoxBody, @@ -38,7 +37,7 @@ enum Direction { } impl Streaming { - pub fn new_response(decoder: D, body: B, status_code: StatusCode) -> Self + pub(crate) fn new_response(decoder: D, body: B, status_code: StatusCode) -> Self where B: Body + Send + 'static, B::Data: Into, @@ -48,7 +47,7 @@ impl Streaming { Self::new(decoder, body, Direction::Response(status_code)) } - pub fn new_empty(decoder: D, body: B) -> Self + pub(crate) fn new_empty(decoder: D, body: B) -> Self where B: Body + Send + 'static, B::Data: Into, @@ -58,7 +57,7 @@ impl Streaming { Self::new(decoder, body, Direction::EmptyResponse) } - pub fn new_request(decoder: D, body: B) -> Self + pub(crate) fn new_request(decoder: D, body: B) -> Self where B: Body + Send + 'static, B::Data: Into, @@ -112,7 +111,7 @@ impl Streaming { // Trailers were not caught during poll_next and thus lets poll for // them manually. let map = - future::poll_fn(|cx| unsafe { Pin::new_unchecked(&mut self.body) }.poll_trailers(cx)) + future::poll_fn(|cx| Pin::new(&mut self.body).poll_trailers(cx)) .await .map_err(|e| Status::from_error(&e))?; @@ -189,8 +188,7 @@ impl Stream for Streaming { None => (), } - // FIXME: Figure out how to verify that this is safe - let chunk = match ready!(unsafe { Pin::new_unchecked(&mut self.body) }.poll_data(cx)) { + let chunk = match ready!(Pin::new(&mut self.body).poll_data(cx)) { Some(Ok(d)) => Some(d), Some(Err(e)) => { let err: crate::Error = e.into(); @@ -205,7 +203,7 @@ impl Stream for Streaming { if let Some(data) = chunk { self.buf.put(data); } else { - // FIXME: get BytesMut to impl `Buf` directlty? + // TODO: get BytesMut to impl `Buf` directlty? let buf1 = (&self.buf[..]).into_buf(); if buf1.has_remaining() { trace!("unexpected EOF decoding stream"); @@ -220,7 +218,7 @@ impl Stream for Streaming { } if let Direction::Response(status) = self.direction { - match ready!(unsafe { Pin::new_unchecked(&mut self.body) }.poll_trailers(cx)) { + match ready!(Pin::new(&mut self.body).poll_trailers(cx)) { Ok(trailer) => { if let Err(e) = crate::status::infer_grpc_status(trailer.as_ref(), status) { return Some(Err(e)).into(); @@ -243,6 +241,6 @@ impl Stream for Streaming { impl fmt::Debug for Streaming { fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { - write!(f, "Streaming") + f.debug_struct("Streaming").finish() } } diff --git a/tonic/src/codec/mod.rs b/tonic/src/codec/mod.rs index dbf5cd0..241afae 100644 --- a/tonic/src/codec/mod.rs +++ b/tonic/src/codec/mod.rs @@ -1,23 +1,40 @@ +//! gRPC encoding and decoding. +//! +//! This module contains the generic `Codec` trait and a protobuf codec +//! based on prost. + mod decode; mod encode; mod prost; pub use self::decode::Streaming; -pub use self::encode::{encode_client, encode_server, EncodeBody}; +pub(crate) use self::encode::{encode_client, encode_server}; pub use self::prost::ProstCodec; +pub use tokio_codec::{Decoder, Encoder}; use crate::Status; -use tokio_codec::{Decoder, Encoder}; +/// Triat that knows how to encode and decode gRPC messages. pub trait Codec { - type Encode; - type Decode; + /// The encodable message. + type Encode: Send + 'static; + /// The decodable message. + type Decode: Send + 'static; - type Encoder: Encoder; - type Decoder: Decoder; + /// The encoder that can encode a message. + type Encoder: Encoder + Send + 'static; + /// The encoder that can decode a message. + type Decoder: Decoder + Send + 'static; + /// The content type of this codec. + /// + /// This should follow the `Content-Type` definition [here]. + /// + /// [here]: https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-HTTP2.md#requests const CONTENT_TYPE: &'static str; + /// Fetch the encoder. fn encoder(&mut self) -> Self::Encoder; + /// Fetch the decoder. fn decoder(&mut self) -> Self::Decoder; } diff --git a/tonic/src/codec/prost.rs b/tonic/src/codec/prost.rs index 14af51d..365f1a0 100644 --- a/tonic/src/codec/prost.rs +++ b/tonic/src/codec/prost.rs @@ -1,16 +1,17 @@ -use super::Codec; +use super::{Codec, Decoder, Encoder}; use crate::{Code, Status}; use bytes::{BufMut, BytesMut}; use prost::Message; use std::marker::PhantomData; -use tokio_codec::{Decoder, Encoder}; +/// A [`Codec`] that implements `application/grpc+proto` via the prost library.. #[derive(Debug, Clone)] pub struct ProstCodec { _pd: PhantomData<(T, U)>, } impl ProstCodec { + /// Create a new codec that knows how to encode `T` and decode `U`. pub fn new() -> Self { Self { _pd: PhantomData } } @@ -18,8 +19,8 @@ impl ProstCodec { impl Codec for ProstCodec where - T: Message, - U: Message + Default, + T: Message + Send + 'static, + U: Message + Default + Send + 'static, { type Encode = T; type Decode = U; @@ -38,6 +39,7 @@ where } } +/// A [`Encoder`] that knows how to encode `T`. pub struct ProstEncoder(PhantomData); impl Encoder for ProstEncoder { @@ -56,6 +58,7 @@ impl Encoder for ProstEncoder { } } +/// A [`Decoder`] that knows how to decode `U`. pub struct ProstDecoder(PhantomData); impl Decoder for ProstDecoder { diff --git a/tonic/src/server/grpc.rs b/tonic/src/server/grpc.rs index b55ee38..f5c11eb 100644 --- a/tonic/src/server/grpc.rs +++ b/tonic/src/server/grpc.rs @@ -7,6 +7,7 @@ use bytes::Bytes; use futures_core::TryStream; use futures_util::{future, stream, TryStreamExt}; use http_body::Body; +use std::fmt; /// A gRPC Server handler. /// @@ -24,10 +25,6 @@ pub struct Grpc { impl Grpc where T: Codec, - T::Decoder: Send + 'static, - T::Decode: Send + Unpin + 'static, - T::Encoder: Send + 'static, - T::Encode: Send + Unpin + 'static, { /// Creates a new gRPC client with the provided [`Codec`]. pub fn new(codec: T) -> Self { @@ -97,8 +94,6 @@ where ) -> http::Response where S: ClientStreamingService, Response = T::Encode>, - T::Decode: Send + 'static, - T::Decoder: Send + 'static, B: Body + Send + 'static, B::Data: Into + Send + 'static, B::Error: Into + Send + 'static, @@ -205,3 +200,9 @@ where } } } + +impl fmt::Debug for Grpc { + fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { + f.debug_struct("Grpc").finish() + } +} diff --git a/tonic/src/server/mod.rs b/tonic/src/server/mod.rs index 319f5ff..7f6deb5 100644 --- a/tonic/src/server/mod.rs +++ b/tonic/src/server/mod.rs @@ -1,4 +1,4 @@ -//! gRPC over HTTP2 server implementation. +//! gRPC server implementation. //! //! This module contains the low level components to build a gRPC server. It //! provides a codec agnostic gRPC server handler.