feat(tonic): support setting timeout for each request

CHANGELOG.md

@@ -6,6 +6,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [0.2.25] - 2023-07-03
+
+### Added
+
+- tonic: Support setting timeout for each request.
+
+### Fixed
+
+- s3: Fix compile error.
+
 ## [0.2.24] - 2023-06-26
 
 ### Fixed

madsim-aws-sdk-s3/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "madsim-aws-sdk-s3"
-version = "0.2.24+0.28"
+version = "0.2.25+0.28"
 edition = "2021"
 authors = ["Kevin Axel <[email protected]>"]
 description = "The s3 simulator on madsim."

madsim-tonic/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "madsim-tonic"
-version = "0.2.21"
+version = "0.2.25"
 edition = "2021"
 authors = ["Runji Wang <[email protected]>"]
 description = "The `tonic` simulator on madsim."

madsim-tonic/src/client.rs

@@ -59,7 +59,7 @@ impl<F: Interceptor> Grpc<crate::transport::Channel, F> {
         M1: Send + Sync + 'static,
         M2: Send + Sync + 'static,
     {
-        let timeout = self.inner.timeout;
+        let timeout = request.timeout().or(self.inner.timeout);
         let future = async move {
             request.append_metadata();
             let request = request.intercept(&mut self.interceptor)?.boxed();
@@ -90,7 +90,7 @@ impl<F: Interceptor> Grpc<crate::transport::Channel, F> {
         M1: Send + Sync + 'static,
         M2: Send + Sync + 'static,
     {
-        let timeout = self.inner.timeout;
+        let timeout = request.timeout().or(self.inner.timeout);
         let future = async move {
             request.append_metadata();
             let request = request.intercept(&mut self.interceptor)?;
@@ -121,7 +121,7 @@ impl<F: Interceptor> Grpc<crate::transport::Channel, F> {
         M1: Send + Sync + 'static,
         M2: Send + Sync + 'static,
     {
-        let timeout = self.inner.timeout;
+        let timeout = request.timeout().or(self.inner.timeout);
         let future = async move {
             request.append_metadata();
             let request = request.intercept(&mut self.interceptor)?.boxed();
@@ -151,7 +151,7 @@ impl<F: Interceptor> Grpc<crate::transport::Channel, F> {
         M1: Send + Sync + 'static,
         M2: Send + Sync + 'static,
     {
-        let timeout = self.inner.timeout;
+        let timeout = request.timeout().or(self.inner.timeout);
         let future = async move {
             request.append_metadata();
             let request = request.intercept(&mut self.interceptor)?;

madsim-tonic/src/sim.rs

@@ -45,6 +45,7 @@ impl<T> AppendMetadata for Result<Response<T>, Status> {
 pub mod codegen {
     use std::any::Any;
     pub use std::net::SocketAddr;
+    use std::time::Duration;
     use tonic::{
         metadata::MetadataMap, service::Interceptor, Extensions, Request, Response, Status,
     };
@@ -60,6 +61,7 @@ pub mod codegen {
     pub type IdentityInterceptor = fn(Request<()>) -> Result<Request<()>, Status>;
 
     pub trait RequestExt<T>: Sized {
+        fn timeout(&self) -> Option<Duration>;
         fn set_remote_addr(&mut self, addr: SocketAddr);
         fn intercept<F: Interceptor>(self, interceptor: &mut F) -> Result<Self, Status>;
         fn boxed(self) -> Request<BoxMessage>
@@ -68,6 +70,22 @@ pub mod codegen {
     }
 
     impl<T> RequestExt<T> for Request<T> {
+        /// Get the timeout of Request.
+        fn timeout(&self) -> Option<Duration> {
+            let s = self.metadata().get("grpc-timeout")?.to_str().unwrap();
+            let (value, unit) = s.split_at(s.len() - 1);
+            let value = value.parse::<u64>().expect("invalid grpc-timeout value");
+            Some(match unit {
+                "H" => Duration::from_secs(value * 60 * 60),
+                "M" => Duration::from_secs(value * 60),
+                "S" => Duration::from_secs(value),
+                "m" => Duration::from_millis(value),
+                "u" => Duration::from_micros(value),
+                "n" => Duration::from_nanos(value),
+                _ => panic!("invalid grpc-timeout unit: {unit}"),
+            })
+        }
+
         /// Set the remote address of Request.
         fn set_remote_addr(&mut self, addr: SocketAddr) {
             let tcp_info: tonic::transport::server::TcpConnectInfo =

tonic-example/tests/test.rs

@@ -393,6 +393,14 @@ async fn request_timeout() {
             let error = client.delay(request()).await.unwrap_err();
             assert_eq!(error.code(), tonic::Code::DeadlineExceeded);
             assert!(t0.elapsed() < Duration::from_secs(2));
+
+            // timeout set on request should override channel timeout
+            let mut req = request();
+            req.set_timeout(Duration::from_secs(5));
+            let t0 = Instant::now();
+            let error = client.delay(req).await.unwrap_err();
+            assert_eq!(error.code(), tonic::Code::DeadlineExceeded);
+            assert!(t0.elapsed() >= Duration::from_secs(5));
         })
         .await
         .unwrap();