Support timeouts with "grpc-timeout" header

tests/integration_tests/tests/timeout.rs

@@ -0,0 +1,52 @@
+use futures_util::FutureExt;
+use integration_tests::pb::{test_client, test_server, Input, Output};
+use std::time::Duration;
+use tokio::sync::oneshot;
+use tonic::{
+    transport::Server,
+    Request, Response, Status,
+};
+
+#[tokio::test]
+async fn cancelation_on_timeout() {
+    struct Svc;
+
+    #[tonic::async_trait]
+    impl test_server::Test for Svc {
+        async fn unary_call(&self, _req: Request<Input>) -> Result<Response<Output>, Status> {
+            // Wait for a time longer than the timeout
+            tokio::time::delay_for(Duration::from_millis(1_000)).await;
+            Ok(Response::new(Output {}))
+        }
+    }
+
+    let svc = test_server::TestServer::new(Svc);
+
+    let (tx, rx) = oneshot::channel::<()>();
+
+    let jh = tokio::spawn(async move {
+        Server::builder()
+            .add_service(svc)
+            .serve_with_shutdown("127.0.0.1:1322".parse().unwrap(), rx.map(drop))
+            .await
+            .unwrap();
+    });
+
+    tokio::time::delay_for(Duration::from_millis(100)).await;
+
+    let mut client = test_client::TestClient::connect("http://127.0.0.1:1322")
+        .await
+        .unwrap();
+
+    let mut req = Request::new(Input {});
+    req.metadata_mut().insert("grpc-timeout", "500m".parse().unwrap());
+
+    let err = client.unary_call(req).await.unwrap_err();
+    assert!(err.message().contains("Timeout expired"));
+    // TODO: Need to return the correct type of code
+    // assert_eq!(err.code(), Code::Cancelled);
+
+    tx.send(()).unwrap();
+
+    jh.await.unwrap();
+}

tonic/Cargo.toml

@@ -65,7 +65,7 @@ async-trait = { version = "0.1.13", optional = true }
 
 # transport
 hyper = { version = "0.13.4", features = ["stream"], optional = true }
-tokio = { version = "0.2.13", features = ["tcp"], optional = true }
+tokio = { version = "0.2.13", features = ["tcp", "time"], optional = true }
 tower = { version = "0.3", optional = true}
 tower-make = { version = "0.3", features = ["connect"] }
 tower-balance =  { version = "0.3", optional = true }

tonic/src/metadata/map.rs

@@ -198,11 +198,10 @@ pub struct OccupiedEntry<'a, VE: ValueEncoding> {
 
 impl MetadataMap {
     // Headers reserved by the gRPC protocol.
-    pub(crate) const GRPC_RESERVED_HEADERS: [&'static str; 8] = [
+    pub(crate) const GRPC_RESERVED_HEADERS: [&'static str; 7] = [
         "te",
         "user-agent",
         "content-type",
-        "grpc-timeout",
         "grpc-message",
         "grpc-encoding",
         "grpc-message-type",

tonic/src/transport/service/connection.rs

@@ -1,5 +1,5 @@
 use super::super::BoxFuture;
-use super::{layer::ServiceBuilderExt, reconnect::Reconnect, AddOrigin, UserAgent};
+use super::{layer::ServiceBuilderExt, reconnect::Reconnect, AddOrigin, Timeout, UserAgent};
 use crate::{body::BoxBody, transport::Endpoint};
 use http::Uri;
 use hyper::client::conn::Builder;
@@ -13,7 +13,6 @@ use tokio::io::{AsyncRead, AsyncWrite};
 use tower::{
     layer::Layer,
     limit::{concurrency::ConcurrencyLimitLayer, rate::RateLimitLayer},
-    timeout::TimeoutLayer,
     util::BoxService,
     ServiceBuilder, ServiceExt,
 };
@@ -53,7 +52,7 @@ impl Connection {
         let stack = ServiceBuilder::new()
             .layer_fn(|s| AddOrigin::new(s, endpoint.uri.clone()))
             .layer_fn(|s| UserAgent::new(s, endpoint.user_agent.clone()))
-            .optional_layer(endpoint.timeout.map(TimeoutLayer::new))
+            .layer_fn(|s| Timeout::new(s, endpoint.timeout))
             .optional_layer(endpoint.concurrency_limit.map(ConcurrencyLimitLayer::new))
             .optional_layer(endpoint.rate_limit.map(|(l, d)| RateLimitLayer::new(l, d)))
             .into_inner();

tonic/src/transport/service/mod.rs

@@ -6,6 +6,7 @@ mod io;
 mod layer;
 mod reconnect;
 mod router;
+mod timeout;
 #[cfg(feature = "tls")]
 mod tls;
 mod user_agent;
@@ -17,6 +18,7 @@ pub(crate) use self::discover::DynamicServiceStream;
 pub(crate) use self::io::ServerIo;
 pub(crate) use self::layer::ServiceBuilderExt;
 pub(crate) use self::router::{Or, Routes};
+pub(crate) use self::timeout::Timeout;
 #[cfg(feature = "tls")]
 pub(crate) use self::tls::{TlsAcceptor, TlsConnector};
 pub(crate) use self::user_agent::UserAgent;