Inital basic not working example

tonic/Cargo.toml

@@ -20,3 +20,9 @@ tower-service = { git = "https://github.com/tower-rs/tower", branch = "std-futur
 tokio-codec = "=0.2.0-alpha.1"
 async-stream = { path = "../../async-stream/async-stream" }
 http-body = { git = "https://github.com/hyperium/http-body", branch = "std-future" }
+
+[dev-dependencies]
+tokio = "=0.2.0-alpha.1"
+tokio-buf = "=0.2.0-alpha.1"
+prost-derive = "0.5"
+tower-h2 = { path = "../tower-h2" }

tonic/examples/server.rs

@@ -0,0 +1,100 @@
+#![feature(async_await, type_alias_impl_trait)]
+
+use std::future::Future;
+use tonic::{server::{UnaryService, Grpc}, Status, Request, Response, body};
+use tower_service::Service;
+use tower_h2::{RecvBody, Server};
+use tokio::net::TcpListener;
+use futures_util::future;
+use std::task::{Context, Poll};
+
+#[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))
+        }
+    }
+}
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let addr = "[::1]:50051".parse().unwrap();
+    let mut bind = TcpListener::bind(&addr)?;
+
+    let mut server = Server::new(MakeSvc, Default::default());
+
+    while let Ok((sock, _addr)) = bind.accept().await {
+        if let Err(e) = sock.set_nodelay(true) {
+            return Err(e.into());
+        }
+
+        if let Err(e) = server.serve(sock).await {
+            println!("H2 ERROR: {}", e);
+        }
+    }
+
+     Ok(())
+}
+
+#[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>>;
+
+    fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        Ok(()).into()
+    }
+
+    fn call(&mut self, req: http::Request<RecvBody>) -> Self::Future {
+        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)
+    }
+}
+
+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)
+    }
+}

tonic/src/body.rs

@@ -1,6 +1,6 @@
 use crate::{Code, Status};
 use bytes::{Bytes, IntoBuf};
-use futures_core::Stream;
+use futures_core::{Stream, TryStream};
 use futures_util::{ready, TryStreamExt};
 use http::HeaderMap;
 use http_body::Body;
@@ -51,12 +51,22 @@ pub struct BoxAsyncBody {
 }
 
 impl BoxAsyncBody {
-    pub fn new<S>(inner: S) -> Self
+    // pub fn new<S>(inner: S) -> Self
+    // where
+    //     S: Stream<Item = Result<crate::body::BytesBuf, Status>> + Send + 'static,
+    // {
+    //     Self {
+    //         inner: Box::pin(inner),
+    //         error: None,
+    //     }
+    // }
+
+    pub fn new_try<S>(inner: S) -> Self
     where
-        S: Stream<Item = Result<crate::body::BytesBuf, Status>> + Send + 'static,
+        S: TryStream<Ok = BytesBuf, Error = Status> + Send + 'static,
     {
         Self {
-            inner: Box::pin(inner),
+            inner: Box::pin(inner.into_stream()),
             error: None,
         }
     }

tonic/src/codec.rs

@@ -1,11 +1,13 @@
 #![allow(dead_code)]
 
-use crate::{body::BytesBuf, Status};
+use crate::{body::BytesBuf, Code, Status};
 use async_stream::stream;
 use bytes::{Buf, BufMut, BytesMut, IntoBuf};
 use futures_core::TryStream;
 use futures_util::{future, TryStreamExt};
 use http_body::Body;
+use prost::Message;
+use std::marker::PhantomData;
 use tokio_codec::{Decoder, Encoder};
 use tracing::{debug, trace};
 
@@ -16,14 +18,13 @@ pub trait Codec {
     type Encoder: Encoder<Item = Self::Encode, Error = Status>;
     type Decoder: Decoder<Item = Self::Decode, Error = Status>;
 
+    const CONTENT_TYPE: &'static str;
+
     fn encoder(&mut self) -> Self::Encoder;
     fn decoder(&mut self) -> Self::Decoder;
 }
 
-pub async fn encode<T, U>(
-    mut encoder: T,
-    mut source: U,
-) -> impl TryStream<Ok = BytesBuf, Error = Status>
+pub fn encode<T, U>(mut encoder: T, mut source: U) -> impl TryStream<Ok = BytesBuf, Error = Status>
 where
     T: Encoder<Error = Status>,
     U: TryStream<Ok = T::Item, Error = Status> + Unpin,
@@ -79,10 +80,10 @@ where
             } else {
                 if buf.has_remaining_mut() {
                     trace!("unexpected EOF decoding stream");
-                    // yield Err(Status::new(
-                    //     Code::Internal,
-                    //     "Unexpected EOF decoding stream.".to_string(),
-                    // ));
+                    yield Err(Status::new(
+                        Code::Internal,
+                        "Unexpected EOF decoding stream.".to_string(),
+                    ));
                 } else {
                     break;
                 }
@@ -153,6 +154,76 @@ where
     Ok(None)
 }
 
+#[derive(Debug, Clone)]
+pub struct ProstCodec<T, U> {
+    _pd: PhantomData<(T, U)>,
+}
+
+impl<T, U> ProstCodec<T, U> {
+    pub fn new() -> Self {
+        Self { _pd: PhantomData }
+    }
+}
+
+impl<T, U> Codec for ProstCodec<T, U>
+where
+    T: Message,
+    U: Message + Default,
+{
+    type Encode = T;
+    type Decode = U;
+
+    type Encoder = ProstEncoder<T>;
+    type Decoder = ProstDecoder<U>;
+
+    const CONTENT_TYPE: &'static str = "application/groc+proto";
+
+    fn encoder(&mut self) -> Self::Encoder {
+        ProstEncoder(PhantomData)
+    }
+
+    fn decoder(&mut self) -> Self::Decoder {
+        ProstDecoder(PhantomData)
+    }
+}
+
+pub struct ProstEncoder<T>(PhantomData<T>);
+
+impl<T: Message> Encoder for ProstEncoder<T> {
+    type Item = T;
+    type Error = Status;
+
+    fn encode(&mut self, item: Self::Item, buf: &mut BytesMut) -> Result<(), Self::Error> {
+        let len = item.encoded_len();
+
+        if buf.remaining_mut() < len {
+            buf.reserve(len);
+        }
+
+        item.encode(buf)
+            .map_err(|_| unreachable!("Message only errors if not enough space"))
+    }
+}
+
+pub struct ProstDecoder<U>(PhantomData<U>);
+
+impl<U: Message + Default> Decoder for ProstDecoder<U> {
+    type Item = U;
+    type Error = Status;
+
+    fn decode(&mut self, buf: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
+        Message::decode(buf.take())
+            .map(Option::Some)
+            .map_err(from_decode_error)
+    }
+}
+
+fn from_decode_error(error: prost::DecodeError) -> crate::Status {
+    // Map Protobuf parse errors to an INTERNAL status code, as per
+    // https://github.com/grpc/grpc/blob/master/doc/statuscodes.md
+    crate::Status::new(crate::Code::Internal, error.to_string())
+}
+
 #[derive(Default)]
 pub struct UnitCodec;
 
@@ -163,6 +234,8 @@ impl Codec for UnitCodec {
     type Encoder = UnitEncoder;
     type Decoder = UnitDecoder;
 
+    const CONTENT_TYPE: &'static str = "()";
+
     fn encoder(&mut self) -> Self::Encoder {
         UnitEncoder
     }

tonic/src/server/mod.rs

@@ -1,12 +1,15 @@
 use crate::{
+    body::{BoxAsyncBody, BytesBuf},
     codec::{self, Codec},
-    Request, Status,
+    Request, Response, Status,
 };
-use futures_core::{Future, TryStream};
+use futures_core::{Future, Stream, 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,
 }
@@ -64,13 +67,12 @@ pub trait StreamingService<RequestStream> {
     fn call(&mut self, request: Request<RequestStream>) -> Self::Future;
 }
 
-type BoxStream<T> = Pin<Box<dyn TryStream<Ok = T, Error = Status> + Send + 'static>>;
-
 impl<T> Grpc<T>
 where
     T: Codec,
     T::Decode: Unpin + 'static,
-    T::Encode: Unpin + 'static,
+    T::Encoder: Send + 'static,
+    T::Encode: Send + Unpin + 'static,
 {
     pub fn new(codec: T) -> Self {
         Self { codec }
@@ -80,7 +82,7 @@ where
         &mut self,
         mut service: S,
         req: http::Request<B>,
-    ) -> http::Response<impl TryStream<Ok = crate::body::BytesBuf, Error = Status>>
+    ) -> http::Response<BoxAsyncBody>
     where
         S: UnaryService<T::Decode, Response = T::Encode>,
         B: Body,
@@ -91,21 +93,22 @@ where
         futures_util::pin_mut!(stream);
         let message = stream.try_next().await.unwrap().unwrap();
         let request = Request::new(message);
-        let response = service.call(request).await.unwrap();
-        let message = response.into_inner();
-        let source = stream::once(future::ok(message));
-        let body = codec::encode(self.codec.encoder(), source).await;
+        let response = service
+            .call(request)
+            .await
+            .map(|r| r.map(|m| stream::once(future::ok(m))));
 
-        http::Response::new(body)
+        self.map_response(response).map(BoxAsyncBody::new_try)
     }
 
     pub async fn server_streaming<S, B>(
         &mut self,
         mut service: S,
         req: http::Request<B>,
-    ) -> http::Response<impl TryStream<Ok = crate::body::BytesBuf, Error = Status>>
+    ) -> http::Response<BoxAsyncBody>
     where
         S: ServerStreamingService<T::Decode, Response = T::Encode>,
+        S::ResponseStream: Send + 'static,
         B: Body,
         B::Error: Into<crate::Error>,
     {
@@ -114,11 +117,9 @@ where
         futures_util::pin_mut!(stream);
         let message = stream.try_next().await.unwrap().unwrap();
         let request = Request::new(message);
-        let response = service.call(request).await.unwrap();
-        let source = response.into_inner();
-        let body = codec::encode(self.codec.encoder(), source).await;
+        let response = service.call(request).await;
 
-        http::Response::new(body)
+        self.map_response(response).map(BoxAsyncBody::new_try)
     }
 
     pub async fn client_streaming<S, B>(
@@ -135,13 +136,13 @@ where
         B::Error: Into<crate::Error> + Send,
     {
         let (_parts, body) = req.into_parts();
-        let stream = codec::decode(self.codec.decoder(), body);
+        let stream = codec::decode(self.codec.decoder(), body).into_stream();
         let stream = Box::pin(stream) as BoxStream<T::Decode>;
         let request = Request::new(stream);
         let response = service.call(request).await.unwrap();
         let message = response.into_inner();
         let source = stream::once(future::ok(message));
-        let body = codec::encode(self.codec.encoder(), source).await;
+        let body = codec::encode(self.codec.encoder(), source);
 
         http::Response::new(body)
     }
@@ -160,13 +161,55 @@ where
         B::Error: Into<crate::Error> + Send,
     {
         let (_parts, body) = req.into_parts();
-        let stream = codec::decode(self.codec.decoder(), body);
+        let stream = codec::decode(self.codec.decoder(), body).into_stream();
         let stream = Box::pin(stream) as BoxStream<T::Decode>;
         let request = Request::new(stream);
         let response = service.call(request).await.unwrap();
         let source = response.into_inner();
-        let body = codec::encode(self.codec.encoder(), source).await;
+        let body = codec::encode(self.codec.encoder(), source);
 
         http::Response::new(body)
     }
+
+    // fn map_request<B>(&mut self, request: http::Request<B>) -> Request<B> {
+    //     Request::from_http(request.map(|b| codec::decode(self.codec.decoder(), b)))
+    // }
+
+    fn map_response<B>(
+        &mut self,
+        response: Result<crate::Response<B>, Status>,
+    ) -> http::Response<BoxStream<BytesBuf>>
+    where
+        B: TryStream<Ok = T::Encode, Error = Status> + Send + Unpin + 'static,
+    {
+        match response {
+            Ok(r) => {
+                let (mut parts, body) = r.into_http().into_parts();
+
+                // Set the content type
+                parts.headers.insert(
+                    http::header::CONTENT_TYPE,
+                    http::header::HeaderValue::from_static(T::CONTENT_TYPE),
+                );
+
+                let body = codec::encode(self.codec.encoder(), body).into_stream();
+
+                let body = Box::pin(body) as BoxStream<BytesBuf>;
+                http::Response::from_parts(parts, body)
+            }
+            Err(status) => {
+                let status = stream::once(future::err(status));
+                let body = codec::encode(self.codec.encoder(), status).into_stream();
+                let (mut parts, _body) = Response::new(()).into_http().into_parts();
+
+                parts.headers.insert(
+                    http::header::CONTENT_TYPE,
+                    http::header::HeaderValue::from_static(T::CONTENT_TYPE),
+                );
+
+                let body = Box::pin(body) as BoxStream<BytesBuf>;
+                http::Response::from_parts(parts, body)
+            }
+        }
+    }
 }

tonic/tests/server.rs

@@ -0,0 +1,55 @@
+#![feature(async_await, type_alias_impl_trait)]
+
+use std::future::Future;
+use std::task::{Context, Poll};
+use tokio_buf::BufStream;
+use tonic::codec::UnitCodec;
+use tonic::server::*;
+use tonic::{Request, Response, Status};
+
+struct SayHello;
+
+impl UnaryService<()> for SayHello {
+    type Response = ();
+    type Future = impl Future<Output = Result<Response<Self::Response>, Status>>;
+
+    fn call(&mut self, _request: Request<()>) -> Self::Future {
+        async move { Ok(Response::new(())) }
+    }
+}
+
+#[tokio::test]
+async fn say_hello() {
+    let codec = UnitCodec::default();
+    let mut grpc = Grpc::new(codec);
+
+    let request = http::Request::new(Body(Vec::new()));
+    grpc.unary(SayHello, request).await;
+}
+
+#[derive(Debug, Default, Clone)]
+struct Body(Vec<u8>);
+
+impl From<Vec<u8>> for Body {
+    fn from(t: Vec<u8>) -> Self {
+        Body(t)
+    }
+}
+
+impl BufStream for Body {
+    type Item = std::io::Cursor<Vec<u8>>;
+    type Error = std::io::Error;
+
+    fn poll_buf(&mut self, _cx: &mut Context<'_>) -> Poll<Option<Result<Self::Item, Self::Error>>> {
+        if self.0.is_empty() {
+            return None.into();
+        }
+
+        use std::{io, mem};
+
+        let bytes = mem::replace(&mut self.0, Default::default());
+        let buf = io::Cursor::new(bytes);
+
+        Some(Ok(buf)).into()
+    }
+}