4.x: Improves support for new gRPC V2 blocking stub calls (#10453)

* Improves support for V2 blocking stub calls. Handles use case of messages in receiving queue that need to be delivered after all data from the server stream is consumed. Some new tests. Use a configurable duration that waits for the listener to request more data when the receiving queue is not empty. If data is received within that period, and more data is in the queue, then the duration resets.

Author: spericas

webclient/grpc/src/main/java/io/helidon/webclient/grpc/GrpcClientCall.java

@@ -22,8 +22,8 @@
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
 import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
 
 import io.helidon.common.buffers.BufferData;
 import io.helidon.webclient.http2.StreamTimeoutException;
@@ -36,9 +36,7 @@
 import static java.lang.System.Logger.Level.ERROR;
 
 /**
- * An implementation of a gRPC call. Expects:
- * <p>
- * start (request | sendMessage)* (halfClose | cancel)
+ * An implementation of a gRPC call.
  *
  * @param <ReqT> request type
  * @param <ResT> response type
@@ -47,7 +45,7 @@ class GrpcClientCall<ReqT, ResT> extends GrpcBaseClientCall<ReqT, ResT> {
     private static final System.Logger LOGGER = System.getLogger(GrpcClientCall.class.getName());
 
     private final ExecutorService executor;
-    private final AtomicInteger messageRequest = new AtomicInteger();
+    private final Semaphore messageRequest = new Semaphore(0);
 
     private final LinkedBlockingQueue<BufferData> sendingQueue = new LinkedBlockingQueue<>();
     private final LinkedBlockingQueue<BufferData> receivingQueue = new LinkedBlockingQueue<>();
@@ -67,7 +65,7 @@ class GrpcClientCall<ReqT, ResT> extends GrpcBaseClientCall<ReqT, ResT> {
     @Override
     public void request(int numMessages) {
         socket().log(LOGGER, DEBUG, "request called %d", numMessages);
-        messageRequest.addAndGet(numMessages);
+        messageRequest.release(numMessages);
         startReadBarrier.countDown();
     }
 
@@ -202,8 +200,24 @@ protected void startStreamingThreads() {
                     socket().log(LOGGER, DEBUG, "[Reading thread] adding bufferData to receiving queue");
                 }
 
-                socket().log(LOGGER, DEBUG, "[Reading thread] closing listener");
-                responseListener().onClose(Status.OK, EMPTY_METADATA);
+                // attempt to drain our receiving queue if permits arrive on time
+                Status status = Status.OK;
+                if (!receivingQueue.isEmpty()) {
+                    Duration waitTime = grpcClient().prototype().protocolConfig().nextRequestWaitTime();
+                    do {
+                        if (messageRequest.tryAcquire(waitTime.toNanos(), TimeUnit.NANOSECONDS)) {
+                            ResT res = toResponse(receivingQueue.remove());
+                            responseListener().onMessage(res);
+                        } else {
+                            socket().log(LOGGER, DEBUG, "[Reading thread] unable to drain receiving queue");
+                            status = Status.CANCELLED;
+                            break;      // wait time expired
+                        }
+                    } while (!receivingQueue.isEmpty());
+                }
+
+                // report onClose call with status
+                responseListener().onClose(status, EMPTY_METADATA);
             } catch (StreamTimeoutException e) {
                 responseListener().onClose(Status.DEADLINE_EXCEEDED, EMPTY_METADATA);
             } catch (Throwable e) {
@@ -236,8 +250,7 @@ private void close() {
 
     private void drainReceivingQueue() {
         socket().log(LOGGER, DEBUG, "[Reading thread] draining receiving queue");
-        while (messageRequest.get() > 0 && !receivingQueue.isEmpty()) {
-            messageRequest.getAndDecrement();
+        while (!receivingQueue.isEmpty() && messageRequest.tryAcquire()) {
             ResT res = toResponse(receivingQueue.remove());
             responseListener().onMessage(res);
         }

webclient/grpc/src/main/java/io/helidon/webclient/grpc/GrpcClientProtocolConfigBlueprint.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2024 Oracle and/or its affiliates.
+ * Copyright (c) 2024, 2025 Oracle and/or its affiliates.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -92,4 +92,19 @@ default String type() {
     @Option.Configured
     @Option.Default("2048")
     int initBufferSize();
+
+    /**
+     * When data has been received from the server but not yet requested by the client
+     * (i.e., listener), the implementation will wait for this duration before signaling
+     * an error. If data is requested and more data is still in the queue, this time
+     * wait restarts until the next request is received. If duration expires, a
+     * status of {@link io.grpc.Status#CANCELLED} is reported in the call to
+     * {@link io.grpc.ClientCall.Listener#onClose(io.grpc.Status, io.grpc.Metadata)}.
+     *
+     * @return duration to wait for the next data request from listener
+     * @see io.grpc.ClientCall.Listener#request(int)
+     */
+    @Option.Configured
+    @Option.Default("PT1S")
+    Duration nextRequestWaitTime();
 }

webclient/tests/grpc/src/test/java/io/helidon/webclient/grpc/tests/GrpcStubTest.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2024 Oracle and/or its affiliates.
+ * Copyright (c) 2024, 2025 Oracle and/or its affiliates.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -31,10 +31,12 @@
 import io.helidon.webclient.grpc.GrpcClient;
 import io.helidon.webserver.WebServer;
 import io.helidon.webserver.testing.junit5.ServerTest;
-import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.Test;
 
+import io.grpc.StatusException;
+import io.grpc.stub.BlockingClientCall;
 import io.grpc.stub.StreamObserver;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
 
 import static org.hamcrest.CoreMatchers.is;
 import static org.hamcrest.MatcherAssert.assertThat;
@@ -119,6 +121,16 @@ void testServerStreamingSplitAsync() throws ExecutionException, InterruptedExcep
         assertThat(res.hasNext(), is(false));
     }
 
+    @Test
+    void testServerStreamingSplitV2() throws InterruptedException, StatusException, TimeoutException {
+        GrpcClient grpcClient = webClient.client(GrpcClient.PROTOCOL);
+        StringServiceGrpc.StringServiceBlockingV2Stub service = StringServiceGrpc.newBlockingV2Stub(grpcClient.channel());
+        BlockingClientCall<?, Strings.StringMessage> call = service.split(newStringMessage("hello world"));
+        assertThat(call.read(100, TimeUnit.MILLISECONDS).getText(), is("hello"));
+        assertThat(call.read(100, TimeUnit.MILLISECONDS).getText(), is("world"));
+        assertThat(call.hasNext(), is(false));
+    }
+
     @Test
     void testClientStreamingJoinAsync() throws ExecutionException, InterruptedException, TimeoutException {
         GrpcClient grpcClient = webClient.client(GrpcClient.PROTOCOL);
@@ -132,6 +144,18 @@ void testClientStreamingJoinAsync() throws ExecutionException, InterruptedExcept
         assertThat(res.getText(), is("hello world"));
     }
 
+    @Test
+    void testClientStreamingJoinV2() throws InterruptedException, TimeoutException, StatusException {
+        GrpcClient grpcClient = webClient.client(GrpcClient.PROTOCOL);
+        StringServiceGrpc.StringServiceBlockingV2Stub service = StringServiceGrpc.newBlockingV2Stub(grpcClient.channel());
+        BlockingClientCall<Strings.StringMessage, Strings.StringMessage> call = service.join();
+        call.write(newStringMessage("hello"));
+        call.write(newStringMessage("world"));
+        call.halfClose();
+        Strings.StringMessage res = call.read(100, TimeUnit.MILLISECONDS);
+        assertThat(res.getText(), is("hello world"));
+    }
+
     @Test
     void testBidirectionalEchoAsync() throws ExecutionException, InterruptedException, TimeoutException {
         GrpcClient grpcClient = webClient.client(GrpcClient.PROTOCOL);
@@ -147,6 +171,26 @@ void testBidirectionalEchoAsync() throws ExecutionException, InterruptedExceptio
         assertThat(res.hasNext(), is(false));
     }
 
+    @Test
+    void testBidirectionalEchoV2() throws InterruptedException, TimeoutException, StatusException, ExecutionException {
+        GrpcClient grpcClient = webClient.client(GrpcClient.PROTOCOL);
+        StringServiceGrpc.StringServiceBlockingV2Stub service = StringServiceGrpc.newBlockingV2Stub(grpcClient.channel());
+        BlockingClientCall<Strings.StringMessage, Strings.StringMessage> call = service.echo();
+        CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
+            try {
+                assertThat(call.read(100, TimeUnit.MILLISECONDS).getText(), is("hello"));
+                assertThat(call.read(100, TimeUnit.MILLISECONDS).getText(), is("world"));
+                assertThat(call.hasNext(), is(false));
+            } catch (InterruptedException | TimeoutException | StatusException e) {
+                throw new RuntimeException(e);
+            }
+        });
+        assertThat(call.write(newStringMessage("hello")), is(true));
+        assertThat(call.write(newStringMessage("world")), is(true));
+        call.halfClose();
+        future.get(TIMEOUT_SECONDS, TimeUnit.SECONDS);
+    }
+
     @Test
     void testBidirectionalEchoAsyncEmpty() throws ExecutionException, InterruptedException, TimeoutException {
         GrpcClient grpcClient = webClient.client(GrpcClient.PROTOCOL);