feat(transport): port router to axum

tonic-build/src/client.rs

@@ -57,7 +57,8 @@ pub fn generate<T: Service>(
             impl<T> #service_ident<T>
             where
                 T: tonic::client::GrpcService<tonic::body::BoxBody>,
-                T::ResponseBody: Body + Send  + 'static,
+                T::ResponseBody: Body<Data = Bytes> + Send  + 'static,
+                <T::ResponseBody as Body>::Error: Into<StdError> + Send + 'static,
                 T::Error: Into<StdError>,
                 <T::ResponseBody as Body>::Error: Into<StdError> + Send,
             {

tonic-build/src/server.rs

@@ -119,7 +119,7 @@ pub fn generate<T: Service>(
                     B::Error: Into<StdError> + Send + 'static,
             {
                 type Response = http::Response<tonic::body::BoxBody>;
-                type Error = Never;
+                type Error = std::convert::Infallible;
                 type Future = BoxFuture<Self::Response, Self::Error>;
 
                 fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {

tonic/Cargo.toml

@@ -32,6 +32,7 @@ tls-roots = ["tls-roots-common", "rustls-native-certs"]
 tls-roots-common = ["tls"]
 tls-webpki-roots = ["tls-roots-common", "webpki-roots"]
 transport = [
+  "axum",
   "h2",
   "hyper",
   "tokio",
@@ -77,6 +78,7 @@ tokio = {version = "1.0.1", features = ["net"], optional = true}
 tokio-stream = "0.1"
 tower = {version = "0.4.7", features = ["balance", "buffer", "discover", "limit", "load", "make", "timeout", "util"], optional = true}
 tracing-futures = {version = "0.2", optional = true}
+axum = {version = "0.3", default_features = false, optional = true}
 
 # rustls
 rustls-native-certs = {version = "0.5", optional = true}

tonic/src/body.rs

@@ -5,6 +5,15 @@ use http_body::Body;
 /// A type erased HTTP body used for tonic services.
 pub type BoxBody = http_body::combinators::UnsyncBoxBody<bytes::Bytes, crate::Status>;
 
+/// Convert a [`http_body::Body`] into a [`BoxBody`].
+pub(crate) fn box_body<B>(body: B) -> BoxBody
+where
+    B: http_body::Body<Data = bytes::Bytes> + Send + 'static,
+    B::Error: Into<crate::Error>,
+{
+    body.map_err(crate::Status::map_error).boxed_unsync()
+}
+
 // this also exists in `crate::codegen` but we need it here since `codegen` has
 // `#[cfg(feature = "codegen")]`.
 /// Create an empty `BoxBody`

tonic/src/codegen.rs

@@ -13,6 +13,7 @@ pub type StdError = Box<dyn std::error::Error + Send + Sync + 'static>;
 #[cfg(feature = "compression")]
 pub use crate::codec::{CompressionEncoding, EnabledCompressionEncodings};
 pub use crate::service::interceptor::InterceptedService;
+pub use bytes::Bytes;
 pub use http_body::Body;
 
 pub type BoxFuture<T, E> = self::Pin<Box<dyn self::Future<Output = Result<T, E>> + Send + 'static>>;
@@ -23,17 +24,6 @@ pub mod http {
     pub use http::*;
 }
 
-#[derive(Debug)]
-pub enum Never {}
-
-impl std::fmt::Display for Never {
-    fn fmt(&self, _: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        match *self {}
-    }
-}
-
-impl std::error::Error for Never {}
-
 pub fn empty_body() -> crate::body::BoxBody {
     http_body::Empty::new()
         .map_err(|err| match err {})

tonic/src/service/interceptor.rs

@@ -2,7 +2,11 @@
 //!
 //! See [`Interceptor`] for more details.
 
-use crate::{request::SanitizeHeaders, Status};
+use crate::{
+    body::{box_body, BoxBody},
+    request::SanitizeHeaders,
+    Status,
+};
 use pin_project::pin_project;
 use std::{
     fmt,
@@ -143,14 +147,16 @@ where
     F: Interceptor,
     S: Service<http::Request<ReqBody>, Response = http::Response<ResBody>>,
     S::Error: Into<crate::Error>,
+    ResBody: http_body::Body<Data = bytes::Bytes> + Send + 'static,
+    ResBody::Error: Into<crate::Error>,
 {
-    type Response = http::Response<ResBody>;
-    type Error = crate::Error;
+    type Response = http::Response<BoxBody>;
+    type Error = S::Error;
     type Future = ResponseFuture<S::Future>;
 
     #[inline]
     fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        self.inner.poll_ready(cx).map_err(Into::into)
+        self.inner.poll_ready(cx)
     }
 
     fn call(&mut self, req: http::Request<ReqBody>) -> Self::Future {
@@ -168,7 +174,7 @@ where
                 let req = req.into_http(uri, SanitizeHeaders::No);
                 ResponseFuture::future(self.inner.call(req))
             }
-            Err(status) => ResponseFuture::error(status),
+            Err(status) => ResponseFuture::status(status),
         }
     }
 }
@@ -197,9 +203,9 @@ impl<F> ResponseFuture<F> {
         }
     }
 
-    fn error(status: Status) -> Self {
+    fn status(status: Status) -> Self {
         Self {
-            kind: Kind::Error(Some(status)),
+            kind: Kind::Status(Some(status)),
         }
     }
 }
@@ -208,22 +214,26 @@ impl<F> ResponseFuture<F> {
 #[derive(Debug)]
 enum Kind<F> {
     Future(#[pin] F),
-    Error(Option<Status>),
+    Status(Option<Status>),
 }
 
 impl<F, E, B> Future for ResponseFuture<F>
 where
     F: Future<Output = Result<http::Response<B>, E>>,
-    E: Into<crate::Error>,
+    B: http_body::Body<Data = bytes::Bytes> + Send + 'static,
+    B::Error: Into<crate::Error>,
 {
-    type Output = Result<http::Response<B>, crate::Error>;
+    type Output = Result<http::Response<BoxBody>, E>;
 
     fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
         match self.project().kind.project() {
-            KindProj::Future(future) => future.poll(cx).map_err(Into::into),
-            KindProj::Error(status) => {
-                let error = status.take().unwrap().into();
-                Poll::Ready(Err(error))
+            KindProj::Future(future) => future
+                .poll(cx)
+                .map(|result| result.map(|res| res.map(box_body)))
+                .map_err(Into::into),
+            KindProj::Status(status) => {
+                let res = status.take().unwrap().to_http().map(box_body);
+                Poll::Ready(Ok(res))
             }
         }
     }