diff --git a/tonic/src/body.rs b/tonic/src/body.rs index 98fe907..c4f4ef7 100644 --- a/tonic/src/body.rs +++ b/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) -> Self - where - S: Stream> + 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(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 { - #[pin] - inner: S, - error: Option, -} - -impl AsyncBody -where - S: Stream>, -{ - pub fn new(inner: S) -> Self { - Self { inner, error: None } - } -} - -impl HttpBody for AsyncBody -where - S: Stream>, -{ - type Data = BytesBuf; - type Error = Status; - - fn is_end_stream(&self) -> bool { - false - } - - fn poll_data( - self: Pin<&mut Self>, - cx: &mut Context<'_>, - ) -> Poll>> { - 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, 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)) - } -} diff --git a/tonic/src/client/grpc.rs b/tonic/src/client/grpc.rs index 5d1d21f..02c94ba 100644 --- a/tonic/src/client/grpc.rs +++ b/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 Grpc { 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(EncodeBody::new_client) + .map(|s| encode_client(codec.encoder(), Box::pin(s))) .map(BoxBody::map_from); let mut request = request.into_http(uri); diff --git a/tonic/src/codec/encode.rs b/tonic/src/codec/encode.rs index edd5b2a..49fcaa9 100644 --- a/tonic/src/codec/encode.rs +++ b/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(mut encoder: T, source: U) -> impl TryStream +pub fn encode_server( + encoder: T, + source: U, +) -> EncodeBody>> +where + T: Encoder, + U: Stream>, +{ + let stream = encode(encoder, source).into_stream(); + EncodeBody::new_server(stream) +} + +pub fn encode_client( + encoder: T, + source: U, +) -> EncodeBody>> +where + T: Encoder, + U: Stream>, +{ + let stream = encode(encoder, source).into_stream(); + EncodeBody::new_client(stream) +} + +fn encode(mut encoder: T, source: U) -> impl TryStream where T: Encoder, U: Stream>, @@ -64,7 +88,7 @@ impl EncodeBody where S: Stream>, { - 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, diff --git a/tonic/src/codec/mod.rs b/tonic/src/codec/mod.rs index afd374a..0802dd7 100644 --- a/tonic/src/codec/mod.rs +++ b/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; diff --git a/tonic/src/server/grpc.rs b/tonic/src/server/grpc.rs index c2fc273..4395706 100644 --- a/tonic/src/server/grpc.rs +++ b/tonic/src/server/grpc.rs @@ -1,15 +1,12 @@ use crate::{ - body::{BoxBody, BytesBuf}, - codec::{decode_request, encode, Codec, Streaming}, + body::BoxBody, + 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 = Pin> + Send + 'static>>; pub struct Grpc { codec: T, @@ -44,8 +41,7 @@ where return self .map_response::>>>(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( @@ -72,15 +68,13 @@ where let request = match self.map_request_unary(req).await { Ok(r) => r, Err(status) => { - return self - .map_response::(Err(status)) - .map(BoxBody::from_stream); + return self.map_response::(Err(status)); } }; let response = service.call(request).await; - self.map_response(response).map(BoxBody::from_stream) + self.map_response(response) } //BoxStream, @@ -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( @@ -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( @@ -161,7 +155,7 @@ where fn map_response( &mut self, response: Result, Status>, - ) -> http::Response> + ) -> http::Response where B: TryStream + 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; - http::Response::from_parts(parts, body) + // let body = Box::pin(body) as BoxStream; + 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; - http::Response::from_parts(parts, body) + http::Response::from_parts(parts, BoxBody::map_from(body)) } } }