Somewhat working routeguide

tonic/src/codec.rs

@@ -6,6 +6,7 @@ use futures_util::{future, StreamExt};
 use http_body::Body;
 use prost::Message;
 use std::marker::PhantomData;
+use std::pin::Pin;
 use tokio_codec::{Decoder, Encoder};
 use tracing::{debug, trace};
 
@@ -22,6 +23,26 @@ pub trait Codec {
     fn decoder(&mut self) -> Self::Decoder;
 }
 
+pub struct Streaming<T> {
+    inner: Pin<Box<dyn Stream<Item = Result<T, Status>> + Send + 'static>>,
+}
+
+impl<T> Streaming<T> {
+    pub fn new(inner: impl Stream<Item = Result<T, Status>> + Send + 'static) -> Self {
+        let inner = Box::pin(inner);
+        Self { inner }
+    }
+}
+
+use std::task::{Context, Poll};
+impl<T> Stream for Streaming<T> {
+    type Item = Result<T, Status>;
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        Pin::new(&mut self.inner).poll_next(cx)
+    }
+}
+
 pub fn encode<T, U>(mut encoder: T, mut source: U) -> impl TryStream<Ok = BytesBuf, Error = Status>
 where
     T: Encoder<Error = Status>,

tonic/src/server/grpc.rs

@@ -1,11 +1,11 @@
 use crate::{
     body::{BoxAsyncBody, BytesBuf},
-    codec::{decode, encode, Codec},
+    codec::{decode, encode, Codec, Streaming},
     server::{ClientStreamingService, ServerStreamingService, StreamingService, UnaryService},
     Code, Request, Response, Status,
 };
 use futures_core::{Stream, TryStream};
-use futures_util::{future, stream, StreamExt, TryStreamExt};
+use futures_util::{future, stream, TryStreamExt};
 use http_body::Body;
 use std::pin::Pin;
 
@@ -83,18 +83,19 @@ where
         self.map_response(response).map(BoxAsyncBody::new_try)
     }
 
+//BoxStream<T::Decode>,
     pub async fn client_streaming<S, B>(
         &mut self,
         mut service: S,
         req: http::Request<B>,
     ) -> http::Response<BoxAsyncBody>
     where
-        S: ClientStreamingService<BoxStream<T::Decode>, Response = T::Encode>,
-        T::Decode: Send,
+        S: ClientStreamingService<Streaming<T::Decode>, Response = T::Encode>,
+        T::Decode: Send + 'static,
         T::Decoder: Send + 'static,
         B: Body + Send + 'static,
-        B::Data: Send,
-        B::Error: Into<crate::Error> + Send,
+        B::Data: Send + 'static,
+        B::Error: Into<crate::Error> + Send + 'static,
     {
         let request = self.map_request_streaming(req);
         let response = service
@@ -110,7 +111,7 @@ where
         req: http::Request<B>,
     ) -> http::Response<BoxAsyncBody>
     where
-        S: StreamingService<BoxStream<T::Decode>, Response = T::Encode> + Send,
+        S: StreamingService<Streaming<T::Decode>, Response = T::Encode> + Send,
         S::ResponseStream: Send + 'static,
         B: Body + Send + 'static,
         B::Data: Send,
@@ -146,7 +147,7 @@ where
     fn map_request_streaming<B>(
         &mut self,
         request: http::Request<B>,
-    ) -> Request<BoxStream<T::Decode>>
+    ) -> Request<Streaming<T::Decode>>
     where
         B: Body + Send + 'static,
         B::Data: Send,
@@ -154,7 +155,7 @@ where
     {
         Request::from_http(
             request.map(|b| {
-                decode(self.codec.decoder(), b).into_stream().boxed() as BoxStream<T::Decode>
+                Streaming::new(decode(self.codec.decoder(), b).into_stream())
             }),
         )
     }

tonic/tests/h2.rs

@@ -0,0 +1,139 @@
+#![feature(async_await, type_alias_impl_trait)]
+
+use futures_util::future;
+use futures_core::Stream;
+use std::pin::Pin;
+use std::future::Future;
+use std::task::{Context, Poll};
+use tokio::net::TcpListener;
+use tonic::{
+    body,
+    server::{Grpc, UnaryService, ClientStreamingService},
+    Request, Response, Status,
+};
+use tower_h2::{RecvBody, Server};
+use tower_service::Service;
+
+#[derive(Clone, PartialEq, prost::Message)]
+pub struct HelloRequest {
+    #[prost(string, tag = "1")]
+    pub name: std::string::String,
+}
+/// The response message containing the greetings
+#[derive(Clone, PartialEq, prost::Message)]
+pub struct HelloReply {
+    #[prost(string, tag = "1")]
+    pub message: std::string::String,
+}
+
+struct SayHello;
+
+impl UnaryService<HelloRequest> for SayHello {
+    type Response = HelloReply;
+    type Future = impl Future<Output = Result<Response<Self::Response>, Status>>;
+
+    fn call(&mut self, request: Request<HelloRequest>) -> Self::Future {
+        async move {
+            println!("REQUEST = {:?}", request);
+
+            let reply = HelloReply {
+                message: "Zomg, it works!".to_string(),
+            };
+
+            Ok(Response::new(reply))
+        }
+    }
+}
+
+struct SayHelloStream;
+
+impl<S> ClientStreamingService<S> for SayHelloStream 
+where S: Stream<Item = Result<HelloRequest, Status>> + Unpin + Send + 'static {
+    type Response = HelloReply;
+    // type Future = impl Future<Output = Result<Response<Self::Response>, Status>>;
+    type Future = Pin<Box<dyn Future<Output = Result<Response<Self::Response>, Status>> + Send + 'static>>;
+
+    fn call(&mut self, _req: Request<S>) -> Self::Future {
+        let fut = async move { 
+            Ok(Response::new(HelloReply { message: "hello".into()})) 
+        };
+        Box::pin(fut)
+    }
+}
+
+#[tokio::test]
+async fn main() {
+    let addr = "[::1]:50051".parse().unwrap();
+    let mut bind = TcpListener::bind(&addr).unwrap();
+
+    let mut server = Server::new(MakeSvc, Default::default());
+
+    while let Ok((sock, _addr)) = bind.accept().await {
+        if let Err(e) = sock.set_nodelay(true) {
+            panic!("error: {}", e);
+        }
+
+        if let Err(e) = server.serve(sock).await {
+            println!("H2 ERROR: {}", e);
+        }
+    }
+}
+
+#[derive(Debug)]
+pub struct Svc;
+
+impl Service<http::Request<RecvBody>> for Svc {
+    type Response = http::Response<body::BoxAsyncBody>;
+    type Error = tonic::error::Never;
+    // type Future = impl Future<Output = Result<Self::Response, Self::Error>>;
+    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>>>>;
+
+    fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        Ok(()).into()
+    }
+
+    fn call(&mut self, req: http::Request<RecvBody>) -> Self::Future {
+        match req.uri().path() {
+            "/greeter.Helloworld/SayHello" => {
+                let fut = async move {
+                    let codec = tonic::codec::ProstCodec::new();
+                    let mut grpc = Grpc::new(codec);
+                    let response = grpc.unary(SayHello, req).await;
+                    Ok(response)
+                };
+
+                Box::pin(fut)
+            }
+
+            "/greeter.Helloworld/SayHelloStreaming" => {
+                let fut = async move {
+                    let codec = tonic::codec::ProstCodec::new();
+                    let mut grpc = Grpc::new(codec);
+                    let response = grpc.client_streaming(SayHelloStream, req).await;
+                    Ok(response)
+                };
+
+                Box::pin(fut)
+            }
+
+            _ => unimplemented!()
+        }
+        
+    }
+}
+
+pub struct MakeSvc;
+
+impl Service<()> for MakeSvc {
+    type Response = Svc;
+    type Error = std::io::Error;
+    type Future = future::Ready<Result<Self::Response, Self::Error>>;
+
+    fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        Ok(()).into()
+    }
+
+    fn call(&mut self, _: ()) -> Self::Future {
+        future::ok(Svc)
+    }
+}