tonic-web: proxy any kind of Service

This allows applying the GrpcWebLayer to any kind of Service, not just
ones that tonic generates. This makes it possible to use tonic-web as a
grpc-web proxy to a gRPC implemented in another language for example.

Author: bouk

tonic-web/src/call.rs

@@ -1,4 +1,3 @@
-use std::error::Error;
 use std::pin::Pin;
 use std::task::{Context, Poll};
 
@@ -9,6 +8,7 @@ use http::{header, HeaderMap, HeaderValue};
 use http_body::{Body, SizeHint};
 use pin_project::pin_project;
 use tonic::Status;
+use tower_http::BoxError;
 
 use self::content_types::*;
 
@@ -62,7 +62,7 @@ pub(crate) struct GrpcWebCall<B> {
     poll_trailers: bool,
 }
 
-impl<B> GrpcWebCall<B> {
+impl<B: Body> GrpcWebCall<B> {
     pub(crate) fn request(inner: B, encoding: Encoding) -> Self {
         Self::new(inner, Direction::Request, encoding)
     }
@@ -108,15 +108,16 @@ impl<B> GrpcWebCall<B> {
     }
 }
 
-impl<B> GrpcWebCall<B>
+impl<B, D> GrpcWebCall<B>
 where
-    B: Body<Data = Bytes>,
-    B::Error: Error,
+    B: Body<Data = D>,
+    B::Error: Into<BoxError> + Send + 'static,
+    D: Buf,
 {
     fn poll_decode(
         mut self: Pin<&mut Self>,
         cx: &mut Context<'_>,
-    ) -> Poll<Option<Result<B::Data, Status>>> {
+    ) -> Poll<Option<Result<Bytes, Status>>> {
         match self.encoding {
             Encoding::Base64 => loop {
                 if let Some(bytes) = self.as_mut().decode_chunk()? {
@@ -139,7 +140,10 @@ where
             },
 
             Encoding::None => match ready!(self.project().inner.poll_data(cx)) {
-                Some(res) => Poll::Ready(Some(res.map_err(internal_error))),
+                Some(res) => Poll::Ready(Some(
+                    res.map(|mut d| d.copy_to_bytes(d.remaining()))
+                        .map_err(internal_error),
+                )),
                 None => Poll::Ready(None),
             },
         }
@@ -148,15 +152,20 @@ where
     fn poll_encode(
         mut self: Pin<&mut Self>,
         cx: &mut Context<'_>,
-    ) -> Poll<Option<Result<B::Data, Status>>> {
+    ) -> Poll<Option<Result<Bytes, Status>>> {
         let mut this = self.as_mut().project();
 
-        if let Some(mut res) = ready!(this.inner.as_mut().poll_data(cx)) {
+        if let Some(res) = ready!(this.inner.as_mut().poll_data(cx)) {
+            let res = res.map(|mut d| d.copy_to_bytes(d.remaining()));
+
+            let bytes: Result<Bytes, <B as Body>::Error>;
             if *this.encoding == Encoding::Base64 {
-                res = res.map(|b| crate::util::base64::STANDARD.encode(b).into())
+                bytes = res.map(|b| crate::util::base64::STANDARD.encode(b).into());
+            } else {
+                bytes = res;
             }
 
-            return Poll::Ready(Some(res.map_err(internal_error)));
+            return Poll::Ready(Some(bytes.map_err(internal_error)));
         }
 
         // this flag is needed because the inner stream never
@@ -182,10 +191,11 @@ where
     }
 }
 
-impl<B> Body for GrpcWebCall<B>
+impl<B, D> Body for GrpcWebCall<B>
 where
-    B: Body<Data = Bytes>,
-    B::Error: Error,
+    B: Body<Data = D>,
+    B::Error: Into<BoxError> + Send + 'static,
+    D: Buf,
 {
     type Data = Bytes;
     type Error = Status;
@@ -216,10 +226,11 @@ where
     }
 }
 
-impl<B> Stream for GrpcWebCall<B>
+impl<B, D> Stream for GrpcWebCall<B>
 where
-    B: Body<Data = Bytes>,
-    B::Error: Error,
+    B: Body<Data = D>,
+    B::Error: Into<BoxError> + Send + 'static,
+    D: Buf,
 {
     type Item = Result<Bytes, Status>;
 
@@ -252,7 +263,8 @@ impl Encoding {
     }
 }
 
-fn internal_error(e: impl std::fmt::Display) -> Status {
+fn internal_error(e: impl Into<BoxError>) -> Status {
+    let e: BoxError = e.into();
     Status::internal(format!("tonic-web: {}", e))
 }
 

tonic-web/src/layer.rs

@@ -4,32 +4,43 @@ use tower_layer::Layer;
 use tower_service::Service;
 
 /// Layer implementing the grpc-web protocol.
-#[derive(Debug, Clone)]
-pub struct GrpcWebLayer {
-    _priv: (),
+#[derive(Debug)]
+pub struct GrpcWebLayer<RespBody> {
+    _markers: std::marker::PhantomData<RespBody>,
 }
 
-impl GrpcWebLayer {
+impl<RespBody> Clone for GrpcWebLayer<RespBody> {
+    fn clone(&self) -> Self {
+        Self {
+            _markers: std::marker::PhantomData,
+        }
+    }
+}
+
+impl<RespBody> GrpcWebLayer<RespBody> {
     /// Create a new grpc-web layer.
-    pub fn new() -> GrpcWebLayer {
-        Self { _priv: () }
+    pub fn new() -> Self {
+        Self {
+            _markers: std::marker::PhantomData,
+        }
     }
 }
 
-impl Default for GrpcWebLayer {
+impl<RespBody> Default for GrpcWebLayer<RespBody> {
     fn default() -> Self {
         Self::new()
     }
 }
 
-impl<S> Layer<S> for GrpcWebLayer
+impl<S, RespBody> Layer<S> for GrpcWebLayer<RespBody>
 where
-    S: Service<http::Request<hyper::Body>, Response = http::Response<BoxBody>>,
-    S: Send + 'static,
+    S: Service<http::Request<BoxBody>, Response = http::Response<RespBody>> + Send + 'static,
     S::Future: Send + 'static,
     S::Error: Into<BoxError> + Send,
+    RespBody: http_body::Body + Send + 'static,
+    RespBody::Error: Into<BoxError> + Send + 'static,
 {
-    type Service = GrpcWebService<S>;
+    type Service = GrpcWebService<S, RespBody>;
 
     fn layer(&self, inner: S) -> Self::Service {
         GrpcWebService::new(inner)

tonic-web/src/lib.rs

@@ -97,6 +97,8 @@
 #![doc(html_root_url = "https://docs.rs/tonic-web/0.9.1")]
 #![doc(issue_tracker_base_url = "https://github.com/hyperium/tonic/issues/")]
 
+use bytes::Buf;
+use http_body::Body;
 pub use layer::GrpcWebLayer;
 pub use service::{GrpcWebService, ResponseFuture};
 
@@ -106,8 +108,11 @@ mod service;
 
 use http::header::HeaderName;
 use std::time::Duration;
-use tonic::{body::BoxBody, server::NamedService};
-use tower_http::cors::{AllowOrigin, CorsLayer};
+use tonic::{body::BoxBody, server::NamedService, Status};
+use tower_http::{
+    cors::{AllowOrigin, CorsLayer},
+    BoxError,
+};
 use tower_layer::Layer;
 use tower_service::Service;
 
@@ -117,17 +122,16 @@ const DEFAULT_EXPOSED_HEADERS: [&str; 3] =
 const DEFAULT_ALLOW_HEADERS: [&str; 4] =
     ["x-grpc-web", "content-type", "x-user-agent", "grpc-timeout"];
 
-type BoxError = Box<dyn std::error::Error + Send + Sync>;
-
 /// Enable a tonic service to handle grpc-web requests with the default configuration.
 ///
 /// You can customize the CORS configuration composing the [`GrpcWebLayer`] with the cors layer of your choice.
-pub fn enable<S>(service: S) -> CorsGrpcWeb<S>
+pub fn enable<S, RespBody>(service: S) -> CorsGrpcWeb<S, RespBody>
 where
-    S: Service<http::Request<hyper::Body>, Response = http::Response<BoxBody>>,
+    S: Service<http::Request<BoxBody>, Response = http::Response<RespBody>>,
     S: Clone + Send + 'static,
     S::Future: Send + 'static,
     S::Error: Into<BoxError> + Send,
+    RespBody: Body,
 {
     let cors = CorsLayer::new()
         .allow_origin(AllowOrigin::mirror_request())
@@ -153,34 +157,52 @@ where
 
 /// A newtype wrapper around [`GrpcWebLayer`] and [`tower_http::cors::CorsLayer`] to allow
 /// `tonic_web::enable` to implement the [`NamedService`] trait.
-#[derive(Debug, Clone)]
-pub struct CorsGrpcWeb<S>(tower_http::cors::Cors<GrpcWebService<S>>);
+#[derive(Debug)]
+pub struct CorsGrpcWeb<S, RespBody>(tower_http::cors::Cors<GrpcWebService<S, RespBody>>);
+
+impl<S: Clone, RespBody> Clone for CorsGrpcWeb<S, RespBody> {
+    fn clone(&self) -> Self {
+        Self(self.0.clone())
+    }
+}
 
-impl<S> Service<http::Request<hyper::Body>> for CorsGrpcWeb<S>
+impl<S, ReqBody, RespBody> Service<http::Request<ReqBody>> for CorsGrpcWeb<S, RespBody>
 where
-    S: Service<http::Request<hyper::Body>, Response = http::Response<BoxBody>>,
+    S: Service<http::Request<BoxBody>, Response = http::Response<RespBody>>,
     S: Clone + Send + 'static,
     S::Future: Send + 'static,
     S::Error: Into<BoxError> + Send,
+    ReqBody: Body + Send + 'static,
+    ReqBody::Error: Into<BoxError> + Send,
+    RespBody: Body + Default + Send + 'static,
+    RespBody::Error: Into<BoxError> + Send + 'static,
 {
-    type Response = S::Response;
-    type Error = S::Error;
-    type Future =
-        <tower_http::cors::Cors<GrpcWebService<S>> as Service<http::Request<hyper::Body>>>::Future;
+    type Response = <tower_http::cors::Cors<GrpcWebService<S, RespBody>> as Service<
+        http::Request<ReqBody>,
+    >>::Response;
+    type Error = <tower_http::cors::Cors<GrpcWebService<S, RespBody>> as Service<
+        http::Request<ReqBody>,
+    >>::Error;
+    type Future = <tower_http::cors::Cors<GrpcWebService<S, RespBody>> as Service<
+        http::Request<ReqBody>,
+    >>::Future;
 
     fn poll_ready(
         &mut self,
         cx: &mut std::task::Context<'_>,
     ) -> std::task::Poll<Result<(), Self::Error>> {
-        self.0.poll_ready(cx)
+        <tower_http::cors::Cors<GrpcWebService<S, RespBody>> as Service<http::Request<ReqBody>>>::poll_ready(
+            &mut self.0,
+            cx,
+        )
     }
 
-    fn call(&mut self, req: http::Request<hyper::Body>) -> Self::Future {
+    fn call(&mut self, req: http::Request<ReqBody>) -> Self::Future {
         self.0.call(req)
     }
 }
 
-impl<S> NamedService for CorsGrpcWeb<S>
+impl<S, RespBody> NamedService for CorsGrpcWeb<S, RespBody>
 where
     S: NamedService,
 {
@@ -205,3 +227,12 @@ pub(crate) mod util {
         );
     }
 }
+
+pub(crate) fn box_body<D: Buf, E: Into<BoxError> + Send>(
+    body: impl Body<Data = D, Error = E> + Send + 'static,
+) -> BoxBody {
+    let bod = body
+        .map_data(|mut d| d.copy_to_bytes(d.remaining()))
+        .map_err(|e| Status::from_error(e.into() as BoxError));
+    bod.boxed_unsync()
+}