Add idle timeout support to HTTP client request

eclipse-vertx · Nov 14, 2023 · 727efa1 · 727efa1
1 parent 1b95677
commit 727efa1
Show file tree

Hide file tree

Showing 3 changed files with 59 additions and 0 deletions.
diff --git a/vertx-grpc-client/src/main/java/io/vertx/grpc/client/GrpcClientRequest.java b/vertx-grpc-client/src/main/java/io/vertx/grpc/client/GrpcClientRequest.java
@@ -95,6 +95,16 @@ public interface GrpcClientRequest<Req, Resp> extends GrpcWriteStream<Req> {
   @Override
   GrpcClientRequest<Req, Resp> drainHandler(@Nullable Handler<Void> handler);
 
+  /**
+   * Sets the amount of time after which, if the request does not return any data within the timeout period,
+   * the request/response is cancelled and the related futures.
+   *
+   * @param timeout the amount of time in milliseconds.
+   * @return a reference to this, so the API can be used fluently
+   */
+  @Fluent
+  GrpcClientRequest<Req, Resp> idleTimeout(long timeout);
+
   /**
    * @return the underlying HTTP connection
    */

diff --git a/vertx-grpc-client/src/main/java/io/vertx/grpc/client/impl/GrpcClientRequestImpl.java b/vertx-grpc-client/src/main/java/io/vertx/grpc/client/impl/GrpcClientRequestImpl.java
@@ -126,6 +126,12 @@ public GrpcClientRequest<Req, Resp> drainHandler(Handler<Void> handler) {
     return this;
   }
 
+  @Override
+  public GrpcClientRequest<Req, Resp> idleTimeout(long timeout) {
+    httpRequest.idleTimeout(timeout);
+    return this;
+  }
+
   @Override public Future<Void> writeMessage(GrpcMessage message) {
     return writeMessage(message, false);
   }

diff --git a/vertx-grpc-client/src/test/java/io/vertx/grpc/client/ClientRequestTest.java b/vertx-grpc-client/src/test/java/io/vertx/grpc/client/ClientRequestTest.java
@@ -465,6 +465,49 @@ public void onCompleted() {
       }));
   }
 
+  @Test
+  public void testIdleTimeout(TestContext should) throws Exception {
+
+    CompletableFuture<Void> cf = new CompletableFuture<>();
+
+    Async done = should.async(2);
+    startServer(new StreamingGrpc.StreamingImplBase() {
+      @Override
+      public StreamObserver<Item> sink(StreamObserver<Empty> responseObserver) {
+        return new StreamObserver<Item>() {
+          @Override
+          public void onNext(Item item) {
+            cf.complete(null);
+          }
+          @Override
+          public void onError(Throwable t) {
+            should.assertEquals(StatusRuntimeException.class, t.getClass());
+            StatusRuntimeException sre = (StatusRuntimeException) t;
+            should.assertEquals(Status.CANCELLED.getCode(), sre.getStatus().getCode());
+            done.countDown();
+          }
+          @Override
+          public void onCompleted() {
+          }
+        };
+      }
+    });
+
+    client = GrpcClient.client(vertx);
+    client.request(SocketAddress.inetSocketAddress(port, "localhost"), StreamingGrpc.getSinkMethod())
+      .onComplete(should.asyncAssertSuccess(callRequest -> {
+        callRequest.write(Item.getDefaultInstance());
+        cf.whenComplete((v, t) -> {
+          long now = System.currentTimeMillis();
+          callRequest.idleTimeout(1000);
+          callRequest.exceptionHandler(err -> {
+            should.assertTrue(System.currentTimeMillis() - now >= 1000);
+            done.countDown();
+          });
+        });
+      }));
+  }
+
   @Test
   public void testCall(TestContext should) throws IOException {