More encoder clean up

tonic/src/body.rs

@@ -1,10 +1,6 @@
 use crate::{Error, Status};
 use bytes::{Buf, Bytes, IntoBuf};
-use futures_core::Stream;
-use futures_util::{ready, TryStreamExt};
-use http::HeaderMap;
 use http_body::Body as HttpBody;
-use pin_project::pin_project;
 use std::pin::Pin;
 use std::task::{Context, Poll};
 
@@ -70,14 +66,6 @@ pub struct BoxBody {
 }
 
 impl BoxBody {
-    pub fn from_stream<S>(s: S) -> Self
-    where
-        S: Stream<Item = Result<crate::body::BytesBuf, Status>> + Send + 'static,
-    {
-        let body = AsyncBody::new(s);
-        Self::map_from(body)
-    }
-
     /// Create a new `BoxBody` mapping item and error to the default types.
     pub fn map_from<B>(inner: B) -> Self
     where
@@ -111,62 +99,3 @@ impl HttpBody for BoxBody {
         HttpBody::poll_trailers(self.inner.as_mut(), cx)
     }
 }
-
-#[pin_project]
-#[derive(Debug)]
-pub struct AsyncBody<S> {
-    #[pin]
-    inner: S,
-    error: Option<Status>,
-}
-
-impl<S> AsyncBody<S>
-where
-    S: Stream<Item = Result<crate::body::BytesBuf, Status>>,
-{
-    pub fn new(inner: S) -> Self {
-        Self { inner, error: None }
-    }
-}
-
-impl<S> HttpBody for AsyncBody<S>
-where
-    S: Stream<Item = Result<crate::body::BytesBuf, Status>>,
-{
-    type Data = BytesBuf;
-    type Error = Status;
-
-    fn is_end_stream(&self) -> bool {
-        false
-    }
-
-    fn poll_data(
-        self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
-    ) -> Poll<Option<Result<Self::Data, Self::Error>>> {
-        let mut self_proj = self.project();
-        match ready!(self_proj.inner.try_poll_next_unpin(cx)) {
-            Some(Ok(d)) => Some(Ok(d)).into(),
-            Some(Err(status)) => {
-                *self_proj.error = Some(status);
-                None.into()
-            }
-            None => None.into(),
-        }
-    }
-
-    fn poll_trailers(
-        self: Pin<&mut Self>,
-        _cx: &mut Context<'_>,
-    ) -> Poll<Result<Option<HeaderMap>, Status>> {
-        // let self_proj = self.project();
-        // let status = if let Some(status) = self_proj.error.take() {
-        //     status
-        // } else {
-        //     Status::new(Code::Ok, "")
-        // };
-
-        // Poll::Ready(Ok(Some(status.to_header_map()?)))
-        Poll::Ready(Ok(None))
-    }
-}

tonic/src/client/grpc.rs

@@ -1,6 +1,6 @@
 use crate::{
     body::{Body, BoxBody},
-    codec::{decode_empty, decode_response, encode, Codec, EncodeBody, Streaming},
+    codec::{decode_empty, decode_response, encode_client, Codec, Streaming},
     Code, GrpcService, Request, Response, Status,
 };
 use futures_core::Stream;
@@ -114,8 +114,7 @@ impl<T> Grpc<T> {
         let uri = Uri::from_parts(parts).expect("path_and_query only is valid Uri");
 
         let request = request
-            .map(|s| encode(codec.encoder(), Box::pin(s)).into_stream())
+            .map(|s| encode_client(codec.encoder(), Box::pin(s)))
-            .map(EncodeBody::new_client)
             .map(BoxBody::map_from);
 
         let mut request = request.into_http(uri);

tonic/src/codec/encode.rs

@@ -9,7 +9,31 @@ use std::pin::Pin;
 use std::task::{Context, Poll};
 use tokio_codec::Encoder;
 
-pub fn encode<T, U>(mut encoder: T, source: U) -> impl TryStream<Ok = BytesBuf, Error = Status>
+pub fn encode_server<T, U>(
+    encoder: T,
+    source: U,
+) -> EncodeBody<impl Stream<Item = Result<BytesBuf, Status>>>
+where
+    T: Encoder<Error = Status>,
+    U: Stream<Item = Result<T::Item, Status>>,
+{
+    let stream = encode(encoder, source).into_stream();
+    EncodeBody::new_server(stream)
+}
+
+pub fn encode_client<T, U>(
+    encoder: T,
+    source: U,
+) -> EncodeBody<impl Stream<Item = Result<BytesBuf, Status>>>
+where
+    T: Encoder<Error = Status>,
+    U: Stream<Item = Result<T::Item, Status>>,
+{
+    let stream = encode(encoder, source).into_stream();
+    EncodeBody::new_client(stream)
+}
+
+fn encode<T, U>(mut encoder: T, source: U) -> impl TryStream<Ok = BytesBuf, Error = Status>
 where
     T: Encoder<Error = Status>,
     U: Stream<Item = Result<T::Item, Status>>,
@@ -64,7 +88,7 @@ impl<S> EncodeBody<S>
 where
     S: Stream<Item = Result<crate::body::BytesBuf, Status>>,
 {
-    pub fn new_client(inner: S) -> Self {
+    pub(crate) fn new_client(inner: S) -> Self {
         Self {
             inner,
             error: None,
@@ -72,7 +96,7 @@ where
         }
     }
 
-    pub fn new_server(inner: S) -> Self {
+    pub(crate) fn new_server(inner: S) -> Self {
         Self {
             inner,
             error: None,

tonic/src/codec/mod.rs

@@ -3,7 +3,7 @@ mod encode;
 mod prost;
 
 pub use self::decode::{decode_empty, decode_request, decode_response, Streaming};
-pub use self::encode::{encode, EncodeBody};
+pub use self::encode::{encode_client, encode_server, EncodeBody};
 pub use self::prost::ProstCodec;
 
 use crate::Status;

tonic/src/server/grpc.rs

@@ -1,15 +1,12 @@
 use crate::{
-    body::{BoxBody, BytesBuf},
+    body::BoxBody,
-    codec::{decode_request, encode, Codec, Streaming},
+    codec::{decode_request, encode_server, Codec, Streaming},
     server::{ClientStreamingService, ServerStreamingService, StreamingService, UnaryService},
     Code, Request, Response, Status,
 };
-use futures_core::{Stream, TryStream};
+use futures_core::TryStream;
 use futures_util::{future, stream, TryStreamExt};
 use http_body::Body;
-use std::pin::Pin;
-
-type BoxStream<T> = Pin<Box<dyn Stream<Item = Result<T, Status>> + Send + 'static>>;
 
 pub struct Grpc<T> {
     codec: T,
@@ -44,8 +41,7 @@ where
                 return self
                     .map_response::<stream::Once<future::Ready<Result<T::Encode, Status>>>>(Err(
                         status,
-                    ))
+                    ));
-                    .map(BoxBody::from_stream);
             }
         };
 
@@ -54,7 +50,7 @@ where
             .await
             .map(|r| r.map(|m| stream::once(future::ok(m))));
 
-        self.map_response(response).map(BoxBody::from_stream)
+        self.map_response(response)
     }
 
     pub async fn server_streaming<S, B>(
@@ -72,15 +68,13 @@ where
         let request = match self.map_request_unary(req).await {
             Ok(r) => r,
             Err(status) => {
-                return self
+                return self.map_response::<S::ResponseStream>(Err(status));
-                    .map_response::<S::ResponseStream>(Err(status))
-                    .map(BoxBody::from_stream);
             }
         };
 
         let response = service.call(request).await;
 
-        self.map_response(response).map(BoxBody::from_stream)
+        self.map_response(response)
     }
 
     //BoxStream<T::Decode>,
@@ -102,7 +96,7 @@ where
             .call(request)
             .await
             .map(|r| r.map(|m| stream::once(future::ok(m))));
-        self.map_response(response).map(BoxBody::from_stream)
+        self.map_response(response)
     }
 
     pub async fn streaming<S, B>(
@@ -119,7 +113,7 @@ where
     {
         let request = self.map_request_streaming(req);
         let response = service.call(request).await;
-        self.map_response(response).map(BoxBody::from_stream)
+        self.map_response(response)
     }
 
     async fn map_request_unary<B>(
@@ -161,7 +155,7 @@ where
     fn map_response<B>(
         &mut self,
         response: Result<crate::Response<B>, Status>,
-    ) -> http::Response<BoxStream<BytesBuf>>
+    ) -> http::Response<BoxBody>
     where
         B: TryStream<Ok = T::Encode, Error = Status> + Send + 'static,
     {
@@ -175,15 +169,15 @@ where
                     http::header::HeaderValue::from_static(T::CONTENT_TYPE),
                 );
 
-                let body = encode(self.codec.encoder(), body.into_stream()).into_stream();
+                let body = encode_server(self.codec.encoder(), body.into_stream());
 
                 // FIXME: try to return impl Trait?
-                let body = Box::pin(body) as BoxStream<BytesBuf>;
+                // let body = Box::pin(body) as BoxStream<BytesBuf>;
-                http::Response::from_parts(parts, body)
+                http::Response::from_parts(parts, BoxBody::map_from(body))
             }
             Err(status) => {
                 let status = stream::once(future::err(status));
-                let body = encode(self.codec.encoder(), status).into_stream();
+                let body = encode_server(self.codec.encoder(), status);
                 let (mut parts, _body) = Response::new(()).into_http().into_parts();
 
                 parts.headers.insert(
@@ -191,8 +185,7 @@ where
                     http::header::HeaderValue::from_static(T::CONTENT_TYPE),
                 );
 
-                let body = Box::pin(body) as BoxStream<BytesBuf>;
+                http::Response::from_parts(parts, BoxBody::map_from(body))
-                http::Response::from_parts(parts, body)
             }
         }
     }