Add buffer support

tonic-interop/Cargo.toml

@@ -25,6 +25,8 @@ hyper = { git = "https://github.com/hyperium/hyper" }
 console = "0.7"
 structopt = "0.2"
 pretty_env_logger = "0.3"
+tracing-fmt = "0.0.1-alpha.1"
+tracing = "0.1"
 
 [build-dependencies]
 tonic-build = { path = "../tonic-build" }

tonic-interop/src/bin/client.rs

@@ -14,7 +14,8 @@ struct Opts {
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
-    pretty_env_logger::init();
+    let sub = tracing_fmt::FmtSubscriber::builder().finish();
+    tracing::subscriber::set_global_default(sub).unwrap();
 
     let matches = Opts::from_args();
 

tonic/Cargo.toml

@@ -26,7 +26,8 @@ hyper = { git = "https://github.com/hyperium/hyper", optional = true}
 
 # tower
 tower-reconnect = { path = "../../tower/tower-reconnect", optional = true }
+tower-buffer = { path = "../../tower/tower-buffer", optional = true }
 
 [features]
 default = ["transport"]
-transport = ["hyper", "tower-reconnect"]
+transport = ["hyper", "tower-reconnect", "tower-buffer"]

tonic/src/transport/client.rs

@@ -11,19 +11,23 @@ use hyper::{Request, Response};
 use std::future::Future;
 use std::pin::Pin;
 use std::task::{Context, Poll};
+use tower_buffer::{future::ResponseFuture, Buffer};
+use tower_service::Service;
 
 type BoxFuture<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>>;
-// #[derive/(Clone)]
+type Inner = Box<
+    dyn Service<
+            Request<BoxBody>,
+            Response = Response<hyper::Body>,
+            Error = crate::Error,
+            Future = BoxFuture<'static, Result<Response<hyper::Body>, crate::Error>>,
+        > + Send
+        + 'static,
+>;
+
+#[derive(Clone)]
 pub struct Client {
-    svc: Box<
+    svc: Buffer<Inner, Request<BoxBody>>,
-        dyn GrpcService<
-                BoxBody,
-                ResponseBody = hyper::Body,
-                Error = crate::Error,
-                Future = BoxFuture<'static, Result<Response<hyper::Body>, crate::Error>>,
-            > + Send
-            + 'static,
-    >,
 }
 
 impl Client {
@@ -35,7 +39,9 @@ impl Client {
         let svc = AddOrigin::new(svc, addr);
         let svc = BoxService::new(svc);
 
-        Ok(Self { svc: Box::new(svc) })
+        let svc = Buffer::new(Box::new(svc) as Inner, 100);
+
+        Ok(Self { svc })
     }
 }
 
@@ -44,19 +50,17 @@ impl GrpcService<BoxBody> for Client {
     type Error = super::Error;
 
     type Future = MapErr<
-        BoxFuture<'static, Result<Response<Self::ResponseBody>, crate::Error>>,
+        ResponseFuture<BoxFuture<'static, Result<Response<Self::ResponseBody>, crate::Error>>>,
         fn(crate::Error) -> super::Error,
     >;
 
     fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        self.svc
+        GrpcService::poll_ready(&mut self.svc, cx)
-            .poll_ready(cx)
             .map_err(|e| super::Error::from((super::ErrorKind::Client, e)))
     }
 
     fn call(&mut self, request: Request<BoxBody>) -> Self::Future {
-        self.svc
+        GrpcService::call(&mut self.svc, request)
-            .call(request)
             .map_err(|e| super::Error::from((super::ErrorKind::Client, e)))
     }
 }