Add client, client codegen, helloworld and routeguide client examples

tonic-examples/Cargo.toml

@@ -10,14 +10,18 @@ edition = "2018"
 name = "helloworld-server"
 path = "src/helloworld/server.rs"
 
-# [[bin]]
+[[bin]]
-# name = "helloworld-client"
+name = "helloworld-client"
-# path = "src/helloworld/client.rs"
+path = "src/helloworld/client.rs"
 
 [[bin]]
 name = "routeguide-server"
 path = "src/routeguide/server.rs"
 
+[[bin]]
+name = "routeguide-client"
+path = "src/routeguide/client.rs"
+
 [dependencies]
 tonic = { path = "../tonic" }
 tower-h2 = { path = "../tower-h2" }
@@ -29,6 +33,7 @@ bytes = "0.4"
 serde_json = "1.0"
 serde = { version = "1.0", features = ["derive"] }
 async-stream = "0.1"
+http = "0.1"
 
 [build-dependencies]
 tonic-build = { path = "../tonic-build" }

tonic-examples/src/helloworld/client.rs

@@ -0,0 +1,32 @@
+#![feature(async_await)]
+
+use tokio::net::TcpStream;
+use tower_h2::{add_origin::AddOrigin, Connection};
+
+pub mod hello_world {
+    include!(concat!(env!("OUT_DIR"), "/helloworld.rs"));
+    tonic::client!(service = "helloworld.Greeter", proto = "self");
+}
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let addr = "[::1]:50051".parse()?;
+    let io = TcpStream::connect(&addr).await?;
+
+    let origin = http::Uri::from_shared(format!("http://{}", addr).into()).unwrap();
+
+    let svc = Connection::handshake(io).await?;
+    let svc = AddOrigin::new(svc, origin);
+
+    let mut client = hello_world::GreeterClient::new(svc);
+
+    let request = tonic::Request::new(hello_world::HelloRequest {
+        name: "hello".into(),
+    });
+
+    let response = client.say_hello(request).await?;
+
+    println!("RESPONSE={:?}", response);
+
+    Ok(())
+}

tonic-examples/src/routeguide/client.rs

@@ -0,0 +1,66 @@
+#![feature(async_await)]
+
+use route_guide::{Point, RouteNote};
+use std::time::{Duration, Instant};
+use tokio::{net::TcpStream, timer::Interval};
+use tonic::Request;
+use tower_h2::{add_origin::AddOrigin, Connection};
+use futures::TryStreamExt;
+
+mod route_guide {
+    include!(concat!(env!("OUT_DIR"), "/routeguide.rs"));
+    tonic::client!(service = "routeguide.RouteGuide", proto = "self");
+}
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let addr = "[::1]:10000".parse()?;
+    let io = TcpStream::connect(&addr).await?;
+
+    let origin = http::Uri::from_shared(format!("http://{}", addr).into()).unwrap();
+
+    let svc = Connection::handshake(io).await?;
+    let svc = AddOrigin::new(svc, origin);
+
+    let mut client = route_guide::RouteGuideClient::new(svc);
+
+    let start = Instant::now();
+
+    let response = client
+        .get_feature(Request::new(Point {
+            latitude: 409146138,
+            longitude: -746188906,
+        }))
+        .await?;
+
+    println!("FEATURE = {:?}", response);
+
+    let outbound = async_stream::try_stream! {
+        let mut interval =  Interval::new_interval(Duration::from_secs(1));
+
+        while let Some(time) = interval.next().await {
+            let elapsed = time.duration_since(start);
+            let note = RouteNote {
+                location: Some(Point {
+                    latitude: 409146138 + elapsed.as_secs() as i32,
+                    longitude: -746188906,
+                }),
+                message: format!("at {:?}", elapsed),
+            };
+
+            yield note;
+        }
+    };
+
+    let request = Request::new(outbound);
+
+    let response = client.route_chat(request).await?;
+
+    let mut inbound = response.into_inner();
+
+    while let Some(note) = inbound.try_next().await? {
+        println!("NOTE = {:?}", note);
+    }
+
+    Ok(())
+}

tonic-examples/src/routeguide/server.rs

@@ -3,19 +3,22 @@
 mod data;
 
 use futures::{Stream, StreamExt};
-use tokio::{net::TcpListener, sync::{mpsc, Lock}};
-use tonic::{Request, Response, Status};
-use tower_h2::Server;
-use std::sync::Arc;
 use std::collections::HashMap;
 use std::hash::{Hash, Hasher};
+use std::sync::Arc;
 use std::time::Instant;
+use tokio::{
+    net::TcpListener,
+    sync::{mpsc, Lock},
+};
+use tonic::{Request, Response, Status};
+use tower_h2::Server;
 
 pub mod routeguide {
     include!(concat!(env!("OUT_DIR"), "/routeguide.rs"));
 }
 
-use routeguide::{Point, Rectangle, Feature, RouteNote, RouteSummary};
+use routeguide::{Feature, Point, Rectangle, RouteNote, RouteSummary};
 
 #[derive(Debug)]
 pub struct RouteGuide {
@@ -51,7 +54,7 @@ impl RouteGuide {
         &self,
         request: Request<Rectangle>,
     ) -> Result<Response<mpsc::Receiver<Result<Feature, Status>>>, Status> {
-               use std::thread;
+        use std::thread;
 
         println!("ListFeatures = {:?}", request);
 
@@ -70,22 +73,20 @@ impl RouteGuide {
             println!(" /// done sending");
         });
 
-
         Ok(Response::new(rx))
     }
 
     pub async fn record_route(
         &self,
         request: Request<impl Stream<Item = Result<Point, Status>>>,
-    ) -> Result<Response<RouteSummary>, Status> 
+    ) -> Result<Response<RouteSummary>, Status> {
-    {
         println!("RecordRoute");
-        
+
         let stream = request.into_inner();
 
-        // Pin the inbound stream to the stack so that we can call next on it        
+        // Pin the inbound stream to the stack so that we can call next on it
         futures::pin_mut!(stream);
-        
+
         let mut summary = RouteSummary::default();
         let mut last_point = None;
         let now = Instant::now();
@@ -127,8 +128,6 @@ impl RouteGuide {
         let stream = request.into_inner();
         let mut state = self.state.clone();
 
-        
-
         let output = async_stream::try_stream! {
             futures::pin_mut!(stream);
 

tonic-macros/src/client.rs

@@ -0,0 +1,92 @@
+use super::{Method, Service};
+use proc_macro2::TokenStream;
+use quote::{format_ident, quote};
+use syn::Path;
+
+pub(crate) fn generate(service: Service, proto: String) -> TokenStream {
+    let mut stream = TokenStream::new();
+
+    for method in &service.methods {
+        let path = format!(
+            "/{}.{}/{}",
+            service.package, service.proto_name, method.proto_name
+        );
+
+        let method = match (method.client_streaming, method.server_streaming) {
+            (false, false) => generate_unary(method, &proto, path),
+            (false, true) => generate_server_streaming(method, &proto, path),
+            (true, false) => generate_client_streaming(method, &proto, path),
+            (true, true) => generate_streaming(method, &proto, path),
+        };
+
+        stream.extend(method);
+    }
+
+    stream
+}
+
+fn generate_unary(method: &Method, proto: &str, path: String) -> TokenStream {
+    let ident = format_ident!("{}", method.name);
+    let request: Path = syn::parse_str(&format!("{}::{}", proto, method.input_type)).unwrap();
+    let response: Path = syn::parse_str(&format!("{}::{}", proto, method.output_type)).unwrap();
+
+    quote! {
+        pub async fn #ident (&mut self, request: tonic::Request<#request>)
+            -> Result<tonic::Response<#response>, tonic::Status> {
+           let codec = tonic::codec::ProstCodec::new();
+           let path = http::uri::PathAndQuery::from_static(#path);
+           self.inner.unary(request, path, codec).await
+        }
+    }
+}
+
+fn generate_server_streaming(method: &Method, proto: &str, path: String) -> TokenStream {
+    let ident = format_ident!("{}", method.name);
+    let request: Path = syn::parse_str(&format!("{}::{}", proto, method.input_type)).unwrap();
+    let response: Path = syn::parse_str(&format!("{}::{}", proto, method.output_type)).unwrap();
+
+    quote! {
+        pub async fn #ident (&mut self, request: tonic::Request<#request>)
+            -> Result<tonic::Response<tonic::codec::Streaming<#response>>, tonic::Status> {
+           let codec = tonic::codec::ProstCodec::new();
+           let path = http::uri::PathAndQuery::from_static(#path);
+           self.inner.server_streaming(request, path, codec).await
+        }
+    }
+}
+
+fn generate_client_streaming(method: &Method, proto: &str, path: String) -> TokenStream {
+    let ident = format_ident!("{}", method.name);
+    let request: Path = syn::parse_str(&format!("{}::{}", proto, method.input_type)).unwrap();
+    let response: Path = syn::parse_str(&format!("{}::{}", proto, method.output_type)).unwrap();
+
+    quote! {
+        pub async fn #ident <S>(&mut self, request: tonic::Request<S>)
+            -> Result<tonic::Response<#response>, tonic::Status>
+            where S: tonic::_codegen::Stream<Item = Result<#request, tonic::Status>> + Send + 'static,
+        {
+           let codec = tonic::codec::ProstCodec::new();
+           let path = http::uri::PathAndQuery::from_static(#path);
+           let request = request.map(|s| Box::pin(s));
+           self.inner.client_streaming(request, path, codec).await
+        }
+    }
+}
+
+fn generate_streaming(method: &Method, proto: &str, path: String) -> TokenStream {
+    let ident = format_ident!("{}", method.name);
+    let request: Path = syn::parse_str(&format!("{}::{}", proto, method.input_type)).unwrap();
+    let response: Path = syn::parse_str(&format!("{}::{}", proto, method.output_type)).unwrap();
+
+    quote! {
+        pub async fn #ident <S>(&mut self, request: tonic::Request<S>)
+            -> Result<tonic::Response<tonic::codec::Streaming<#response>>, tonic::Status>
+            where S: tonic::_codegen::Stream<Item = Result<#request, tonic::Status>> + Send + 'static,
+        {
+           let codec = tonic::codec::ProstCodec::new();
+           let path = http::uri::PathAndQuery::from_static(#path);
+           let request = request.map(|s| Box::pin(s));
+           self.inner.streaming(request, path, codec).await
+        }
+    }
+}

tonic-macros/src/lib.rs

@@ -3,12 +3,44 @@
 
 extern crate proc_macro;
 
+mod client;
 mod service;
 
 use proc_macro::TokenStream;
+use quote::quote;
 use serde::Deserialize;
 use syn::{AttributeArgs, ItemImpl};
 
+#[proc_macro]
+pub fn client(attr: TokenStream) -> TokenStream {
+    let args = syn::parse_macro_input!(attr as AttributeArgs);
+    let (service, proto_path) = load_service(args);
+
+    let service_ident = quote::format_ident!("{}Client", service.name);
+    let methods = client::generate(service, proto_path);
+
+    let output = quote! {
+        pub struct #service_ident <T> {
+            inner: tonic::client::Grpc<T>,
+        }
+
+        impl<T> #service_ident <T>
+        where T: tonic::GrpcService<tonic::body::BoxAsyncBody>,
+              T::ResponseBody: tonic::body::Body + tonic::_codegen::HttpBody + Send + 'static,
+              <T::ResponseBody as tonic::_codegen::HttpBody>::Error: Into<tonic::error::Error> + Send,
+              <T::ResponseBody as tonic::_codegen::HttpBody>::Data: Send, {
+            pub fn new(inner: T) -> Self {
+                let inner = tonic::client::Grpc::new(inner);
+                Self { inner }
+            }
+
+            #methods
+        }
+    };
+
+    TokenStream::from(output)
+}
+
 #[proc_macro_attribute]
 pub fn server(attr: TokenStream, item: TokenStream) -> TokenStream {
     let mut original = item.clone();

tonic-macros/src/service.rs

@@ -272,7 +272,7 @@ fn generate_client_streaming(
         struct #service_ident(pub std::sync::Arc<#service_impl>);
 
         impl<S> tonic::server::ClientStreamingService<S> for #service_ident
-        where S: Stream<Item = Result<#request, Status>> + Unpin + Send + 'static {
+        where S: tonic::_codegen::Stream<Item = Result<#request, Status>> + Unpin + Send + 'static {
             type Response = #response;
             type Future = BoxFuture<tonic::Response<Self::Response>, tonic::Status>;
 

tonic/src/body.rs

@@ -1,14 +1,63 @@
-use crate::{Code, Status};
+use crate::{Code, Error, Status};
-use bytes::{Bytes, IntoBuf};
+use bytes::{Buf, Bytes, IntoBuf};
 use futures_core::{Stream, TryStream};
 use futures_util::{ready, TryStreamExt};
 use http::HeaderMap;
-use http_body::Body;
+use http_body::Body as HttpBody;
 use std::pin::Pin;
 use std::task::{Context, Poll};
 
 pub type BytesBuf = <Bytes as IntoBuf>::Buf;
 
+pub trait Body: sealed::Sealed {
+    type Data: Buf;
+    type Error: Into<Error>;
+
+    fn is_end_stream(&self) -> bool;
+
+    fn poll_data(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<Self::Data, Self::Error>>>;
+
+    fn poll_trailers(
+        &mut self,
+        cx: &mut Context<'_>,
+    ) -> Poll<Result<Option<http::HeaderMap>, Self::Error>>;
+}
+
+impl<T> Body for T
+where
+    T: HttpBody,
+    T::Error: Into<Error>,
+{
+    type Data = T::Data;
+    type Error = T::Error;
+
+    fn is_end_stream(&self) -> bool {
+        HttpBody::is_end_stream(self)
+    }
+
+    fn poll_data(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<Self::Data, Self::Error>>> {
+        HttpBody::poll_data(self, cx)
+    }
+
+    fn poll_trailers(
+        &mut self,
+        cx: &mut Context<'_>,
+    ) -> Poll<Result<Option<http::HeaderMap>, Self::Error>> {
+        HttpBody::poll_trailers(self, cx)
+    }
+}
+
+impl<T> sealed::Sealed for T
+where
+    T: HttpBody,
+    T::Error: Into<Error>,
+{
+}
+
+mod sealed {
+    pub trait Sealed {}
+}
+
 pub struct BoxBody {
     inner: Box<dyn Body<Data = BytesBuf, Error = Status> + Send>,
 }
@@ -25,7 +74,7 @@ impl BoxBody {
     }
 }
 
-impl Body for BoxBody {
+impl HttpBody for BoxBody {
     type Data = BytesBuf;
     type Error = Status;
 
@@ -72,7 +121,7 @@ impl BoxAsyncBody {
     }
 }
 
-impl Body for BoxAsyncBody {
+impl HttpBody for BoxAsyncBody {
     type Data = BytesBuf;
     type Error = Status;
 
@@ -114,7 +163,7 @@ where
     }
 }
 
-impl<S> Body for AsyncBody<S>
+impl<S> HttpBody for AsyncBody<S>
 where
     S: Stream<Item = Result<crate::body::BytesBuf, Status>> + Unpin,
 {

tonic/src/client/grpc.rs

@@ -0,0 +1,157 @@
+use crate::{
+    body::{Body, BoxAsyncBody},
+    codec::{decode, encode, Codec, Streaming},
+    Code, GrpcService, Request, Response, Status,
+};
+use futures_core::Stream;
+use futures_util::{future, stream, TryStreamExt};
+use http::{
+    header::{HeaderValue, CONTENT_TYPE, TE},
+    uri::{Parts, PathAndQuery, Uri},
+};
+use http_body::Body as HttpBody;
+
+pub struct Grpc<T> {
+    inner: T,
+}
+
+impl<T> Grpc<T> {
+    pub fn new(inner: T) -> Self {
+        Self { inner }
+    }
+
+    pub async fn unary<M1, M2, C>(
+        &mut self,
+        request: Request<M1>,
+        path: PathAndQuery,
+        codec: C,
+    ) -> Result<Response<M2>, Status>
+    where
+        T: GrpcService<BoxAsyncBody>,
+        T::ResponseBody: Body + HttpBody + Send + 'static,
+        <T::ResponseBody as HttpBody>::Error: Into<crate::Error> + Send,
+        <T::ResponseBody as HttpBody>::Data: Send,
+        C: Codec<Encode = M1, Decode = M2>,
+        C::Encoder: Send + 'static,
+        C::Decoder: Send + 'static,
+        M1: Send + 'static,
+        M2: Send + Unpin + 'static,
+    {
+        let request = request.map(|m| stream::once(future::ok(m)));
+        self.client_streaming(request, path, codec).await
+    }
+
+    pub async fn client_streaming<S, M1, M2, C>(
+        &mut self,
+        request: Request<S>,
+        path: PathAndQuery,
+        codec: C,
+    ) -> Result<Response<M2>, Status>
+    where
+        T: GrpcService<BoxAsyncBody>,
+        T::ResponseBody: Body + HttpBody + Send + 'static,
+        <T::ResponseBody as HttpBody>::Error: Into<crate::Error> + Send,
+        <T::ResponseBody as HttpBody>::Data: Send,
+        S: Stream<Item = Result<M1, Status>> + Send + 'static,
+        C: Codec<Encode = M1, Decode = M2>,
+        C::Encoder: Send + 'static,
+        C::Decoder: Send + 'static,
+        M1: Send,
+        M2: Send + Unpin + 'static,
+    {
+        let response = self.streaming(request, path, codec).await?;
+
+        // TODO: use response to parts
+        let mut body = response.into_inner();
+        let message = body
+            .try_next()
+            .await?
+            .ok_or(Status::new(Code::Internal, "Missing response message."))?;
+
+        Ok(Response::new(message))
+    }
+
+    pub async fn server_streaming<M1, M2, C>(
+        &mut self,
+        request: Request<M1>,
+        path: PathAndQuery,
+        codec: C,
+    ) -> Result<Response<Streaming<M2>>, Status>
+    where
+        T: GrpcService<BoxAsyncBody>,
+        T::ResponseBody: Body + HttpBody + Send + 'static,
+        <T::ResponseBody as HttpBody>::Error: Into<crate::Error> + Send,
+        <T::ResponseBody as HttpBody>::Data: Send,
+        C: Codec<Encode = M1, Decode = M2>,
+        C::Encoder: Send + 'static,
+        C::Decoder: Send + 'static,
+        M1: Send + 'static,
+        M2: Send + Unpin + 'static,
+    {
+        let request = request.map(|m| stream::once(future::ok(m)));
+        self.streaming(request, path, codec).await
+    }
+
+    pub async fn streaming<S, M1, M2, C>(
+        &mut self,
+        request: Request<S>,
+        path: PathAndQuery,
+        mut codec: C,
+    ) -> Result<Response<Streaming<M2>>, Status>
+    where
+        T: GrpcService<BoxAsyncBody>,
+        T::ResponseBody: Body + HttpBody + Send + 'static,
+        <T::ResponseBody as HttpBody>::Error: Into<crate::Error> + Send,
+        <T::ResponseBody as HttpBody>::Data: Send,
+        S: Stream<Item = Result<M1, Status>> + Send + 'static,
+        C: Codec<Encode = M1, Decode = M2>,
+        C::Encoder: Send + 'static,
+        C::Decoder: Send + 'static,
+        M1: Send,
+        M2: Send + Unpin + 'static,
+    {
+        let mut parts = Parts::default();
+        parts.path_and_query = Some(path);
+
+        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)))
+            .map(BoxAsyncBody::new_try);
+
+        let mut request = request.into_http(uri);
+
+        // Add the gRPC related HTTP headers
+        request
+            .headers_mut()
+            .insert(TE, HeaderValue::from_static("trailers"));
+
+        // Set the content type
+        // TODO: Don't hard code this here
+        let content_type = <C as Codec>::CONTENT_TYPE;
+        request
+            .headers_mut()
+            .insert(CONTENT_TYPE, HeaderValue::from_static(content_type));
+
+        let response = self
+            .inner
+            .call(request)
+            .await
+            .map_err(|err| Status::from_error(&*(err.into())))?;
+
+        let status_code = response.status();
+        let trailers_only_status = Status::from_header_map(response.headers());
+
+        if let Some(status) = trailers_only_status {
+            if status.code() != Code::Ok {
+                return Err(status);
+            }
+        }
+
+        let response = response
+            .map(|b| decode(codec.decoder(), b).into_stream())
+            .map(Streaming::new);
+
+        Ok(Response::from_http(response))
+    }
+}

tonic/src/client/mod.rs

@@ -0,0 +1,3 @@
+mod grpc;
+
+pub use self::grpc::Grpc;

tonic/src/codec.rs

@@ -1,314 +0,0 @@
-use crate::{body::BytesBuf, Code, Status};
-use async_stream::stream;
-use bytes::{Buf, BufMut, BytesMut, IntoBuf};
-use futures_core::{Stream, TryStream};
-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};
-
-pub trait Codec {
-    type Encode;
-    type Decode;
-
-    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 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>,
-    U: Stream<Item = Result<T::Item, Status>> + Unpin,
-{
-    stream! {
-        let mut buf = BytesMut::with_capacity(1024);
-
-        loop {
-            match source.next().await {
-                Some(Ok(item)) => {
-                    buf.reserve(5);
-                    unsafe {
-                        buf.advance_mut(5);
-                    }
-                    encoder.encode(item, &mut buf).map_err(drop).unwrap();
-
-                    // now that we know length, we can write the header
-                    let len = buf.len() - 5;
-                    assert!(len <= ::std::u32::MAX as usize);
-                    {
-                        let mut cursor = ::std::io::Cursor::new(&mut buf[..5]);
-                        cursor.put_u8(0); // byte must be 0, reserve doesn't auto-zero
-                        cursor.put_u32_be(len as u32);
-                    }
-
-                    yield Ok(buf.split_to(len + 5).freeze().into_buf());
-                },
-                Some(Err(status)) => yield Err(status),
-                None => break,
-            }
-        }
-    }
-}
-
-pub fn decode<T, B>(
-    mut decoder: T,
-    mut source: B,
-) -> impl TryStream<Ok = T::Item, Error = Status> + 'static
-where
-    T: Decoder<Error = Status> + 'static,
-    T::Item: Unpin + 'static,
-    B: Body + 'static,
-    B::Error: Into<crate::Error>,
-{
-    stream! {
-        let mut buf = BytesMut::with_capacity(1024);
-        let mut state = State::ReadHeader;
-
-        loop {
-            // TODO: use try_stream! and ?
-            if let Some(item) = decode_chunk(&mut decoder, &mut buf, &mut state).unwrap() {
-                yield Ok(item);
-            }
-
-            let chunk = match future::poll_fn(|cx| source.poll_data(cx)).await {
-                Some(Ok(d)) => Some(d),
-                Some(Err(e)) => {
-                    let err = e.into();
-                    debug!("decoder inner stream error: {:?}", err);
-                    let status = Status::from_error(&*err);
-                    yield Err(status);
-                    break;
-                },
-                None => None,
-            };
-
-            if let Some(data)= chunk {
-                buf.put(data);
-            } else {
-                if buf.has_remaining_mut() {
-                    trace!("unexpected EOF decoding stream");
-                    yield Err(Status::new(
-                        Code::Internal,
-                        "Unexpected EOF decoding stream.".to_string(),
-                    ));
-                } else {
-                    break;
-                }
-            }
-
-            // TODO: poll_trailers for Response status code
-        }
-    }
-}
-
-fn decode_chunk<T>(
-    decoder: &mut T,
-    buf1: &mut BytesMut,
-    state: &mut State,
-) -> Result<Option<T::Item>, Status>
-where
-    T: Decoder<Error = Status>,
-{
-    let mut buf = (&buf1[..]).into_buf();
-
-    if let State::ReadHeader = state {
-        println!("reading header");
-        if buf.remaining() < 5 {
-            return Ok(None);
-        }
-
-        let is_compressed = match buf.get_u8() {
-            0 => false,
-            1 => {
-                trace!("message compressed, compression not supported yet");
-                return Err(crate::Status::new(
-                    crate::Code::Unimplemented,
-                    "Message compressed, compression not supported yet.".to_string(),
-                ));
-            }
-            f => {
-                trace!("unexpected compression flag");
-                return Err(crate::Status::new(
-                    crate::Code::Internal,
-                    format!("Unexpected compression flag: {}", f),
-                ));
-            }
-        };
-        let len = buf.get_u32_be() as usize;
-
-        *state = State::ReadBody {
-            compression: is_compressed,
-            len,
-        }
-    }
-
-    if let State::ReadBody { len, .. } = state {
-        println!("reading body");
-        if buf.remaining() < *len {
-            return Ok(None);
-        }
-
-        // advance past the header
-        buf1.advance(5);
-
-        match decoder.decode(buf1) {
-            Ok(Some(msg)) => {
-                *state = State::ReadHeader;
-                return Ok(Some(msg));
-            }
-            Ok(None) => return Ok(None),
-            Err(e) => {
-                return Err(e);
-            }
-        }
-    }
-
-    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
-    Status::new(Code::Internal, error.to_string())
-}
-
-#[derive(Default)]
-pub struct UnitCodec;
-
-impl Codec for UnitCodec {
-    type Encode = ();
-    type Decode = ();
-
-    type Encoder = UnitEncoder;
-    type Decoder = UnitDecoder;
-
-    const CONTENT_TYPE: &'static str = "()";
-
-    fn encoder(&mut self) -> Self::Encoder {
-        UnitEncoder
-    }
-
-    fn decoder(&mut self) -> Self::Decoder {
-        UnitDecoder
-    }
-}
-
-pub struct UnitEncoder;
-
-impl Encoder for UnitEncoder {
-    type Item = ();
-    type Error = crate::Status;
-
-    fn encode(&mut self, _item: Self::Item, _buf: &mut BytesMut) -> Result<(), Self::Error> {
-        unimplemented!()
-    }
-}
-
-pub struct UnitDecoder;
-
-impl Decoder for UnitDecoder {
-    type Item = ();
-    type Error = Status;
-
-    fn decode(&mut self, _buf: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
-        Ok(Some(()))
-    }
-}
-
-#[derive(Debug)]
-enum State {
-    ReadHeader,
-    ReadBody { compression: bool, len: usize },
-}

tonic/src/codec/decode.rs

@@ -0,0 +1,155 @@
+use crate::{Code, Status};
+use bytes::{Buf, BufMut, BytesMut, IntoBuf};
+use futures_core::{Stream, TryStream};
+use futures_util::future;
+use http::StatusCode;
+use http_body::Body;
+use std::pin::Pin;
+use tokio_codec::Decoder;
+use tracing::{debug, trace};
+
+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)
+    }
+}
+
+#[derive(Debug)]
+enum State {
+    ReadHeader,
+    ReadBody { compression: bool, len: usize },
+}
+
+enum Direction {
+    Request,
+    Response(StatusCode),
+    EmptyResponse,
+}
+
+pub fn decode<T, B>(
+    mut decoder: T,
+    mut source: B,
+) -> impl TryStream<Ok = T::Item, Error = Status> + 'static
+where
+    T: Decoder<Error = Status> + 'static,
+    T::Item: Unpin + 'static,
+    B: Body + 'static,
+    B::Error: Into<crate::Error>,
+{
+    async_stream::stream! {
+        let mut buf = BytesMut::with_capacity(1024 * 1024);
+        let mut state = State::ReadHeader;
+
+        loop {
+            // TODO: use try_stream! and ?
+            if let Some(item) = decode_chunk(&mut decoder, &mut buf, &mut state).unwrap() {
+                yield Ok(item);
+            }
+
+            let chunk = match future::poll_fn(|cx| source.poll_data(cx)).await {
+                Some(Ok(d)) => Some(d),
+                Some(Err(e)) => {
+                    let err = e.into();
+                    debug!("decoder inner stream error: {:?}", err);
+                    let status = Status::from_error(&*err);
+                    yield Err(status);
+                    break;
+                },
+                None => None,
+            };
+
+            if let Some(data) = chunk {
+                buf.put(data);
+            } else {
+                if buf.has_remaining_mut() {
+                    trace!("unexpected EOF decoding stream");
+                    yield Err(Status::new(
+                        Code::Internal,
+                        "Unexpected EOF decoding stream.".to_string(),
+                    ));
+                } else {
+                    break;
+                }
+            }
+
+            // TODO: poll_trailers for Response status code
+        }
+    }
+}
+
+fn decode_chunk<T>(
+    decoder: &mut T,
+    buf1: &mut BytesMut,
+    state: &mut State,
+) -> Result<Option<T::Item>, Status>
+where
+    T: Decoder<Error = Status>,
+{
+    let mut buf = (&buf1[..]).into_buf();
+
+    if let State::ReadHeader = state {
+        if buf.remaining() < 5 {
+            return Ok(None);
+        }
+
+        let is_compressed = match buf.get_u8() {
+            0 => false,
+            1 => {
+                trace!("message compressed, compression not supported yet");
+                return Err(crate::Status::new(
+                    crate::Code::Unimplemented,
+                    "Message compressed, compression not supported yet.".to_string(),
+                ));
+            }
+            f => {
+                trace!("unexpected compression flag");
+                return Err(crate::Status::new(
+                    crate::Code::Internal,
+                    format!("Unexpected compression flag: {}", f),
+                ));
+            }
+        };
+        let len = buf.get_u32_be() as usize;
+
+        *state = State::ReadBody {
+            compression: is_compressed,
+            len,
+        }
+    }
+
+    if let State::ReadBody { len, .. } = state {
+        if buf.remaining() < *len {
+            return Ok(None);
+        }
+
+        // advance past the header
+        buf1.advance(5);
+
+        match decoder.decode(buf1) {
+            Ok(Some(msg)) => {
+                *state = State::ReadHeader;
+                return Ok(Some(msg));
+            }
+            Ok(None) => return Ok(None),
+            Err(e) => {
+                return Err(e);
+            }
+        }
+    }
+
+    Ok(None)
+}

tonic/src/codec/encode.rs

@@ -0,0 +1,40 @@
+use crate::{body::BytesBuf, Status};
+use bytes::{BufMut, BytesMut, IntoBuf};
+use futures_core::{Stream, TryStream};
+use futures_util::StreamExt;
+use tokio_codec::Encoder;
+
+pub fn encode<T, U>(mut encoder: T, mut source: U) -> impl TryStream<Ok = BytesBuf, Error = Status>
+where
+    T: Encoder<Error = Status>,
+    U: Stream<Item = Result<T::Item, Status>> + Unpin,
+{
+    async_stream::stream! {
+        let mut buf = BytesMut::with_capacity(1024);
+
+        loop {
+            match source.next().await {
+                Some(Ok(item)) => {
+                    buf.reserve(5);
+                    unsafe {
+                        buf.advance_mut(5);
+                    }
+                    encoder.encode(item, &mut buf).map_err(drop).unwrap();
+
+                    // now that we know length, we can write the header
+                    let len = buf.len() - 5;
+                    assert!(len <= std::u32::MAX as usize);
+                    {
+                        let mut cursor = std::io::Cursor::new(&mut buf[..5]);
+                        cursor.put_u8(0); // byte must be 0, reserve doesn't auto-zero
+                        cursor.put_u32_be(len as u32);
+                    }
+
+                    yield Ok(buf.split_to(len + 5).freeze().into_buf());
+                },
+                Some(Err(status)) => yield Err(status),
+                None => break,
+            }
+        }
+    }
+}

tonic/src/codec/mod.rs

@@ -0,0 +1,23 @@
+mod decode;
+mod encode;
+mod prost;
+
+pub use self::decode::{decode, Streaming};
+pub use self::encode::encode;
+pub use self::prost::ProstCodec;
+
+use crate::Status;
+use tokio_codec::{Decoder, Encoder};
+
+pub trait Codec {
+    type Encode;
+    type Decode;
+
+    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;
+}

tonic/src/codec/prost.rs

@@ -0,0 +1,76 @@
+use super::Codec;
+use crate::{Code, Status};
+use bytes::{BufMut, BytesMut};
+use prost::Message;
+use std::marker::PhantomData;
+use tokio_codec::{Decoder, Encoder};
+
+#[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
+    Status::new(Code::Internal, error.to_string())
+}

tonic/src/error.rs

@@ -1,7 +1,7 @@
 use std::fmt;
 
 #[allow(dead_code)]
-pub(crate) type Error = Box<dyn std::error::Error + Send + Sync>;
+pub type Error = Box<dyn std::error::Error + Send + Sync>;
 
 #[derive(Debug)]
 #[allow(dead_code)]

tonic/src/lib.rs

@@ -4,6 +4,7 @@
 //! gRPC implementation
 
 pub mod body;
+pub mod client;
 pub mod codec;
 #[doc(hidden)]
 pub mod error;
@@ -12,13 +13,15 @@ pub mod server;
 
 mod request;
 mod response;
+mod service;
 mod status;
 
 pub use body::{BoxAsyncBody, BoxBody};
 pub use request::Request;
 pub use response::Response;
+pub use service::GrpcService;
 pub use status::{Code, Status};
-pub use tonic_macros::server;
+pub use tonic_macros::{client, server};
 
 pub(crate) use error::Error;
 
@@ -35,7 +38,9 @@ pub trait GrpcInnerService<Request> {
 #[doc(hidden)]
 
 pub mod _codegen {
+    pub use futures_core::Stream;
     pub use futures_util::future::{ok, Ready};
+    pub use http_body::Body as HttpBody;
     pub use std::future::Future;
     pub use std::pin::Pin;
     pub use std::task::{Context, Poll};

tonic/src/server/grpc.rs

@@ -83,7 +83,7 @@ where
         self.map_response(response).map(BoxAsyncBody::new_try)
     }
 
-//BoxStream<T::Decode>,
+    //BoxStream<T::Decode>,
     pub async fn client_streaming<S, B>(
         &mut self,
         mut service: S,
@@ -154,9 +154,7 @@ where
         B::Error: Into<crate::Error> + Send,
     {
         Request::from_http(
-            request.map(|b| {
+            request.map(|b| Streaming::new(decode(self.codec.decoder(), b).into_stream())),
-                Streaming::new(decode(self.codec.decoder(), b).into_stream())
-            }),
         )
     }
 

tonic/src/service.rs

@@ -0,0 +1,37 @@
+use crate::body::Body;
+use http::{Request, Response};
+use http_body::Body as HttpBody;
+use std::future::Future;
+use std::task::{Context, Poll};
+use tower_service::Service;
+
+pub trait GrpcService<ReqBody> {
+    type ResponseBody: Body + HttpBody;
+    type Error: Into<crate::Error>;
+
+    type Future: Future<Output = Result<Response<Self::ResponseBody>, Self::Error>>;
+
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>>;
+
+    fn call(&mut self, request: Request<ReqBody>) -> Self::Future;
+}
+
+impl<T, ReqBody, ResBody> GrpcService<ReqBody> for T
+where
+    T: Service<Request<ReqBody>, Response = Response<ResBody>>,
+    T::Error: Into<crate::Error>,
+    ResBody: Body + HttpBody,
+    <ResBody as HttpBody>::Error: Into<crate::Error>,
+{
+    type ResponseBody = ResBody;
+    type Error = T::Error;
+    type Future = T::Future;
+
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        Service::poll_ready(self, cx)
+    }
+
+    fn call(&mut self, request: Request<ReqBody>) -> Self::Future {
+        Service::call(self, request)
+    }
+}

tonic/tests/h2.rs

@@ -1,14 +1,14 @@
 #![feature(async_await, type_alias_impl_trait)]
 
-use futures_util::future;
 use futures_core::Stream;
-use std::pin::Pin;
+use futures_util::future;
 use std::future::Future;
+use std::pin::Pin;
 use std::task::{Context, Poll};
 use tokio::net::TcpListener;
 use tonic::{
     body,
-    server::{Grpc, UnaryService, ClientStreamingService},
+    server::{ClientStreamingService, Grpc, UnaryService},
     Request, Response, Status,
 };
 use tower_h2::{RecvBody, Server};
@@ -47,15 +47,20 @@ impl UnaryService<HelloRequest> for SayHello {
 
 struct SayHelloStream;
 
-impl<S> ClientStreamingService<S> for SayHelloStream 
+impl<S> ClientStreamingService<S> for SayHelloStream
-where S: Stream<Item = Result<HelloRequest, Status>> + Unpin + Send + 'static {
+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>>;
+    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 { 
+        let fut = async move {
-            Ok(Response::new(HelloReply { message: "hello".into()})) 
+            Ok(Response::new(HelloReply {
+                message: "hello".into(),
+            }))
         };
         Box::pin(fut)
     }
@@ -116,9 +121,8 @@ impl Service<http::Request<RecvBody>> for Svc {
                 Box::pin(fut)
             }
 
-            _ => unimplemented!()
+            _ => unimplemented!(),
         }
-        
     }
 }
 

tonic/tests/server.rs

@@ -1,13 +1,13 @@
 #![feature(async_await, type_alias_impl_trait)]
 
+use futures_core::Stream;
 use std::future::Future;
+use std::pin::Pin;
 use std::task::{Context, Poll};
 use tokio_buf::BufStream;
 use tonic::codec::ProstCodec;
 use tonic::server::*;
 use tonic::{Request, Response, Status};
-use std::pin::Pin;
-use futures_core::Stream;
 
 #[derive(Clone, PartialEq, prost::Message)]
 pub struct HelloRequest {
@@ -28,20 +28,31 @@ impl UnaryService<HelloRequest> for SayHello {
     type Future = impl Future<Output = Result<Response<Self::Response>, Status>>;
 
     fn call(&mut self, _request: Request<HelloRequest>) -> Self::Future {
-        async move { Ok(Response::new(HelloReply { message: "hello".into()})) }
+        async move {
+            Ok(Response::new(HelloReply {
+                message: "hello".into(),
+            }))
+        }
     }
 }
 
 struct SayHelloStream;
 
-impl<S> ClientStreamingService<S> for SayHelloStream 
+impl<S> ClientStreamingService<S> for SayHelloStream
-where S: Stream<Item = Result<HelloRequest, Status>> + Unpin + Send + 'static {
+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>>;
+    type Future =
+        Pin<Box<dyn Future<Output = Result<Response<Self::Response>, Status>> + Send + 'static>>;
 
     fn call(&mut self, _: Request<S>) -> Self::Future {
-        let fut = async move { Ok(Response::new(HelloReply { message: "hello".into()})) };
+        let fut = async move {
+            Ok(Response::new(HelloReply {
+                message: "hello".into(),
+            }))
+        };
         Box::pin(fut)
     }
 }

tower-h2/src/add_origin.rs

@@ -0,0 +1,48 @@
+use http::{Request, Uri};
+use std::task::{Context, Poll};
+use tower_service::Service;
+
+#[derive(Debug)]
+pub struct AddOrigin<T> {
+    inner: T,
+    origin: Uri,
+}
+
+impl<T> AddOrigin<T> {
+    pub fn new(inner: T, origin: Uri) -> Self {
+        Self { inner, origin }
+    }
+}
+
+impl<T, ReqBody> Service<Request<ReqBody>> for AddOrigin<T>
+where
+    T: Service<Request<ReqBody>>,
+{
+    type Response = T::Response;
+    type Error = T::Error;
+    type Future = T::Future;
+
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.inner.poll_ready(cx)
+    }
+
+    fn call(&mut self, req: Request<ReqBody>) -> Self::Future {
+        // Split the request into the head and the body.
+        let (mut head, body) = req.into_parts();
+
+        // Split the request URI into parts.
+        let mut uri: http::uri::Parts = head.uri.into();
+        let set_uri = self.origin.clone().into_parts();
+
+        // Update the URI parts, setting hte scheme and authority
+        uri.scheme = Some(set_uri.scheme.expect("expected scheme").clone());
+        uri.authority = Some(set_uri.authority.expect("expected authority").clone());
+
+        // Update the the request URI
+        head.uri = http::Uri::from_parts(uri).expect("valid uri");
+
+        let request = Request::from_parts(head, body);
+
+        self.inner.call(request)
+    }
+}

tower-h2/src/client.rs

@@ -1,6 +1,6 @@
 use crate::{buf::SendBuf, flush::Flush, recv_body::RecvBody};
 use futures_util::{future, FutureExt, TryFutureExt};
-use h2::{client::SendRequest, RecvStream};
+use h2::client::SendRequest;
 use http::{Request, Response};
 use http_body::Body;
 use std::future::Future;

tower-h2/src/flush.rs

@@ -96,7 +96,7 @@ where
                             match ready!(self.h2.poll_capacity(cx)) {
                                 Some(Ok(0)) => {}
                                 Some(Ok(_)) => break,
-                                Some(Err(e)) => return panic!("error {:?}", e),
+                                Some(Err(e)) => panic!("error {:?}", e),
                                 None => {
                                     debug!("connection closed early");
                                     // The error shouldn't really matter at this

tower-h2/src/lib.rs

@@ -3,6 +3,8 @@
 #[macro_use]
 extern crate log;
 
+pub mod add_origin;
+
 mod buf;
 mod client;
 mod error;

tower-h2/src/recv_body.rs

@@ -1,5 +1,4 @@
 use bytes::{Buf, Bytes, BytesMut};
-use futures_core::Stream;
 use futures_util::TryStreamExt;
 use http_body::Body;
 use std::task::{Context, Poll};