diff --git a/changes.md b/changes.md
index 3674696dc2..28fa95112c 100644
--- a/changes.md
+++ b/changes.md
@@ -95,15 +95,18 @@ Changes log
bounded messages/calls, one-shot SSE responses, and request cancellation. Sessions,
listening GET streams, replay and Ikanos revision compatibility are not supported.
Borrowed HTTP connectors are not stopped by the MCP client.
- - Added experimental SDK-native gRPC unary and bounded server-streaming hosting on a
- borrowed Restlet/Netty HTTP/2 listener. Generated and dynamic services can share the
+ - Added experimental SDK-native gRPC hosting of unary, server-streaming,
+ client-streaming and bidirectional methods on a borrowed Restlet/Netty HTTP/2 listener.
+ Request messages are read as the application requests them, within the queue budget,
+ so an unread request pushes back on the client, while responses stream out; a
+ bidirectional call exchanges messages both ways over one HTTP/2 stream. Generated and dynamic services can share the
listener with REST routes. grpc-java owns service dispatch, framing and status semantics;
Restlet owns HTTP I/O and finite message, queue, concurrency and call-duration budgets.
Cancellation and response completion terminate each logical transport exactly once;
admission remains held until response transmission or cancellation. Adapter shutdown
leaves the borrowed listener and supplied callback executor running and reports timeout
- explicitly. Client-streaming, bidirectional streaming, TLS configuration and physical
- gRPC connection controls are not supported by the server adapter.
+ explicitly. TLS configuration and physical gRPC connection controls are not supported
+ by the server adapter.
- Added an experimental SDK-native gRPC unary client on a borrowed Restlet HTTP/2
connector, with explicit plaintext opt-in, bounded messages and call admission.
Regression tests cover authority mapping, cancellation before and after headers,
diff --git a/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/GrpcServerAdapter.java b/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/GrpcServerAdapter.java
index 8f698ed6dd..81f2acb29c 100644
--- a/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/GrpcServerAdapter.java
+++ b/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/GrpcServerAdapter.java
@@ -30,9 +30,11 @@
/**
* Socketless grpc-java server on a borrowed Restlet HTTP/2 listener.
*
- *
Supports unary requests and server-streamed responses, with finite message, queue, concurrency
- * and duration budgets. Does not implement TLS or physical connection controls. The host
- * application owns the HTTP listener and any supplied callback executor.
+ *
Supports unary, server-streaming, client-streaming and bidirectional methods, with finite
+ * message, queue, concurrency and duration budgets. Request messages are read as the application
+ * requests them, and responses are sent as they are produced, so a bidirectional call can exchange
+ * messages both ways over one HTTP/2 stream. Does not implement TLS or physical connection
+ * controls. The host application owns the HTTP listener and any supplied callback executor.
*
*
Register services before {@link #start()}, attach them to a router with {@link #attachTo}, and
* configure the host listener for plaintext HTTP/2. This adapter is single-use. Stopping it cancels
@@ -93,6 +95,12 @@ public int getMaxMessageBytes() {
}
}
+ /**
+ * Sets the per-call queue budget, for each direction: response bytes waiting for the network,
+ * and request bytes read but not yet consumed by the messages the application requested.
+ *
+ * @param value At least 16384 bytes.
+ */
public void setMaxQueuedBytes(int value) {
synchronized (lock) {
configurable();
@@ -201,11 +209,11 @@ public synchronized void start() throws Exception {
configurable();
for (var service : List.copyOf(services)) {
for (var method : service.getMethods()) {
- var type = method.getMethodDescriptor().getType();
- if (type != io.grpc.MethodDescriptor.MethodType.UNARY
- && type != io.grpc.MethodDescriptor.MethodType.SERVER_STREAMING) {
+ if (method.getMethodDescriptor().getType()
+ == io.grpc.MethodDescriptor.MethodType.UNKNOWN) {
throw new UnsupportedOperationException(
- "Only unary and server-streaming methods are supported");
+ "Methods of unknown type are not supported: "
+ + method.getMethodDescriptor().getFullMethodName());
}
}
}
diff --git a/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletServerHost.java b/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletServerHost.java
index 21a5af88ca..42fcde9a1a 100644
--- a/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletServerHost.java
+++ b/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletServerHost.java
@@ -323,6 +323,10 @@ private final class Call implements ServerTransport {
private final InternalLogId id = InternalLogId.allocate(Call.class, "Restlet logical call");
private final ArrayDeque queue = new ArrayDeque<>();
+
+ /** Guards {@link #unread}, and wakes the request reader. */
+ private final Object inbound = new Object();
+
private final CountDownLatch head = new CountDownLatch(1);
private final AtomicBoolean finished = new AtomicBoolean();
private final AtomicLong operations = new AtomicLong();
@@ -330,8 +334,14 @@ private final class Call implements ServerTransport {
/** 0 before the application callbacks, 1 while they may run, 2 once released. */
private final AtomicInteger callbackState = new AtomicInteger();
- /** The network read or write in progress, 0 if none. */
- private final AtomicLong blocked = new AtomicLong();
+ /**
+ * The network read in progress, 0 if none. Reads and writes run concurrently, so each has
+ * its own marker: the end of one must not hide the other from {@link #abortIfStuck}.
+ */
+ private final AtomicLong reading = new AtomicLong();
+
+ /** The network write in progress, 0 if none. */
+ private final AtomicLong writing = new AtomicLong();
private ServerTransportListener transportListener;
private SdkStream stream;
@@ -339,6 +349,9 @@ private final class Call implements ServerTransport {
private Metadata trailers;
private int queued;
+ /** Request bytes handed to the SDK but not consumed yet; guarded by {@link #inbound}. */
+ private long unread;
+
/** Bytes framed for the latest message, from the frame that started it. */
private int open;
@@ -410,19 +423,12 @@ private void begin() throws IOException, InterruptedException {
transportListener.streamCreated(stream, method, headers);
stream.state.onStreamAllocated();
});
- byte[] bytes;
- try (var input = request.getEntity().getStream()) {
- bytes = input == null ? new byte[0] : read(input);
- }
- if (bytes.length > maxBytes + 5) {
- cancel(Status.RESOURCE_EXHAUSTED.withDescription("Request byte budget exceeded"));
+ InputStream input = request.getEntity().getStream();
+ if (input == null) {
+ deliver(new byte[0], true);
} else {
- serial.execute(
- () -> {
- if (!reported) {
- stream.state.inboundDataReceived(ReadableBuffers.wrap(bytes), true);
- }
- });
+ // Read apart, so that the response can start while the client still sends.
+ Thread.ofVirtual().name("restlet-grpc-server-read").start(() -> readAll(input));
}
if (!head.await(duration.toNanos(), TimeUnit.NANOSECONDS)) {
cancel(Status.DEADLINE_EXCEEDED.withDescription("Response headers timed out"));
@@ -449,12 +455,73 @@ private Attributes transportAttributes() {
return result.build();
}
- private byte[] read(InputStream input) throws IOException {
- enterIo();
- try {
- return input.readNBytes(maxBytes + 6);
- } finally {
- blocked.set(0);
+ /**
+ * Hands the request body to the SDK as it arrives. Bytes handed over but not yet consumed
+ * by the SDK's deframer, which only reads the messages the application requested, stay
+ * within the queue budget: a slow consumer stops the reads, and so the client.
+ */
+ private void readAll(InputStream input) {
+ try (input) {
+ byte[] chunk = new byte[16384];
+ for (; ; ) {
+ int room;
+ synchronized (inbound) {
+ while (unread >= queueBytes && !finished.get()) {
+ inbound.wait();
+ }
+ if (finished.get()) {
+ return;
+ }
+ room = (int) Math.min(chunk.length, queueBytes - unread);
+ }
+ int count;
+ enterIo(reading);
+ try {
+ count = input.read(chunk, 0, room);
+ } finally {
+ reading.set(0);
+ }
+ if (count < 0) {
+ deliver(new byte[0], true);
+ return;
+ }
+ if (count > 0) {
+ synchronized (inbound) {
+ unread += count;
+ }
+ deliver(java.util.Arrays.copyOf(chunk, count), false);
+ }
+ }
+ } catch (InterruptedException _) {
+ Thread.currentThread().interrupt();
+ cancel(Status.CANCELLED.withDescription("gRPC request read interrupted"));
+ } catch (IOException | RuntimeException e) {
+ synchronized (queue) {
+ if (terminal) {
+ // The response ended, so the server closed the unread request body; the
+ // delivery or disconnect signal settles the call, not this read.
+ return;
+ }
+ }
+ cancel(Status.CANCELLED.withDescription("gRPC request read failed").withCause(e));
+ }
+ }
+
+ /** Hands request bytes to the SDK, in {@link #serial} after the stream's creation. */
+ private void deliver(byte[] bytes, boolean end) {
+ serial.execute(
+ () -> {
+ if (!reported) {
+ stream.state.inboundDataReceived(ReadableBuffers.wrap(bytes), end);
+ }
+ });
+ }
+
+ /** Frees inbound budget once the SDK's deframer consumed request bytes. */
+ private void consumed(int count) {
+ synchronized (inbound) {
+ unread -= count;
+ inbound.notifyAll();
}
}
@@ -476,12 +543,12 @@ public void write(OutputStream output) throws IOException {
}
queued -= data.length;
}
- enterIo();
+ enterIo(writing);
try {
output.write(data);
output.flush();
} finally {
- blocked.set(0);
+ writing.set(0);
}
int sent = data.length;
serial.execute(() -> stream.state.onSentBytes(sent));
@@ -499,8 +566,8 @@ public void write(OutputStream output) throws IOException {
}
/** Marks the start of a network read or write, which only the peer may otherwise end. */
- private void enterIo() {
- blocked.set(operations.incrementAndGet());
+ private void enterIo(AtomicLong marker) {
+ marker.set(operations.incrementAndGet());
if (finished.get()) {
abortIfStuck();
}
@@ -512,8 +579,9 @@ private void enterIo() {
* elsewhere was already woken by the end of the call.
*/
private void abortIfStuck() {
- long operation = blocked.get();
- if (operation == 0) {
+ long read = reading.get();
+ long write = writing.get();
+ if (read == 0 && write == 0) {
return;
}
Thread.startVirtualThread(
@@ -524,7 +592,8 @@ private void abortIfStuck() {
Thread.currentThread().interrupt();
return;
}
- if (blocked.get() == operation) {
+ if ((read != 0 && reading.get() == read)
+ || (write != 0 && writing.get() == write)) {
request.abort();
}
});
@@ -584,6 +653,9 @@ private boolean claim() {
if (timeout != null) {
timeout.cancel(false);
}
+ synchronized (inbound) {
+ inbound.notifyAll();
+ }
return true;
}
@@ -750,7 +822,9 @@ public void runOnTransportThread(Runnable task) {
}
@Override
- public void bytesRead(int count) {}
+ public void bytesRead(int count) {
+ call.consumed(count);
+ }
@Override
public void deframeFailed(Throwable cause) {
diff --git a/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcServerAdapterTestCase.java b/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcServerAdapterTestCase.java
index 434bfa6dd6..da4d3b61b4 100644
--- a/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcServerAdapterTestCase.java
+++ b/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcServerAdapterTestCase.java
@@ -78,6 +78,16 @@ class GrpcServerAdapterTestCase {
.setType(MethodDescriptor.MethodType.SERVER_STREAMING)
.setFullMethodName("review.Server/Watch")
.build();
+ private static final MethodDescriptor UPLOAD =
+ ECHO.toBuilder()
+ .setType(MethodDescriptor.MethodType.CLIENT_STREAMING)
+ .setFullMethodName("review.Server/Upload")
+ .build();
+ private static final MethodDescriptor CHAT =
+ ECHO.toBuilder()
+ .setType(MethodDescriptor.MethodType.BIDI_STREAMING)
+ .setFullMethodName("review.Server/Chat")
+ .build();
private final GrpcServerAdapter adapter = new GrpcServerAdapter();
private Server http;
private ManagedChannel channel;
@@ -369,33 +379,297 @@ void requestAndResponseMessageBudgetsAreEnforced() throws Exception {
}
@Test
- void unsupportedStreamingRegistrationFailsExplicitly() {
+ void unknownMethodTypesFailExplicitly() {
adapter.getServices()
.add(
ServerServiceDefinition.builder("review.Server")
.addMethod(
ECHO.toBuilder()
- .setType(
- MethodDescriptor.MethodType
- .CLIENT_STREAMING)
+ .setType(MethodDescriptor.MethodType.UNKNOWN)
.build(),
- ServerCalls.asyncClientStreamingCall(
- out ->
- new io.grpc.stub.StreamObserver<
- StringValue>() {
- @Override
- public void onNext(StringValue value) {}
-
- @Override
- public void onError(Throwable cause) {}
-
- @Override
- public void onCompleted() {}
- }))
+ ServerCalls.asyncUnaryCall(
+ (StringValue request,
+ io.grpc.stub.StreamObserver
+ out) -> out.onCompleted()))
.build());
assertThrows(UnsupportedOperationException.class, adapter::start);
}
+ /** Counts the messages and characters of an uploaded stream. */
+ private static ServerServiceDefinition upload() {
+ return ServerServiceDefinition.builder("review.Server")
+ .addMethod(
+ UPLOAD,
+ ServerCalls.asyncClientStreamingCall(
+ out ->
+ new io.grpc.stub.StreamObserver() {
+ private int messages;
+ private long characters;
+
+ @Override
+ public void onNext(StringValue value) {
+ messages++;
+ characters += value.getValue().length();
+ }
+
+ @Override
+ public void onError(Throwable cause) {}
+
+ @Override
+ public void onCompleted() {
+ out.onNext(
+ StringValue.of(
+ messages + ":" + characters));
+ out.onCompleted();
+ }
+ }))
+ .build();
+ }
+
+ @Test
+ void clientStreamingReceivesEveryMessage() throws Exception {
+ start(upload());
+ var result = new java.util.concurrent.CompletableFuture();
+ var requests =
+ ClientCalls.asyncClientStreamingCall(
+ channel.newCall(
+ UPLOAD,
+ CallOptions.DEFAULT.withDeadlineAfter(15, TimeUnit.SECONDS)),
+ observer(result));
+ for (int i = 0; i < 200; i++) {
+ requests.onNext(StringValue.of("x".repeat(4096)));
+ }
+ requests.onCompleted();
+ assertEquals("200:" + (200 * 4096), result.get(15, TimeUnit.SECONDS));
+ }
+
+ /** Each reply is received before the next request is sent, over one HTTP/2 stream. */
+ @Test
+ void bidirectionalCallsExchangeMessagesBothWays() throws Exception {
+ start(
+ ServerServiceDefinition.builder("review.Server")
+ .addMethod(
+ CHAT,
+ ServerCalls.asyncBidiStreamingCall(
+ out ->
+ new io.grpc.stub.StreamObserver() {
+ @Override
+ public void onNext(StringValue value) {
+ out.onNext(
+ StringValue.of(
+ "re:" + value.getValue()));
+ }
+
+ @Override
+ public void onError(Throwable cause) {
+ /* Not used by this test. */
+ }
+
+ @Override
+ public void onCompleted() {
+ out.onCompleted();
+ }
+ }))
+ .build());
+ var replies = new java.util.concurrent.LinkedBlockingQueue();
+ var done = new java.util.concurrent.CompletableFuture();
+ var requests =
+ ClientCalls.asyncBidiStreamingCall(
+ channel.newCall(
+ CHAT, CallOptions.DEFAULT.withDeadlineAfter(15, TimeUnit.SECONDS)),
+ new io.grpc.stub.StreamObserver() {
+ @Override
+ public void onNext(StringValue value) {
+ replies.add(value.getValue());
+ }
+
+ @Override
+ public void onError(Throwable cause) {
+ done.completeExceptionally(cause);
+ }
+
+ @Override
+ public void onCompleted() {
+ done.complete(null);
+ }
+ });
+ for (int i = 0; i < 5; i++) {
+ requests.onNext(StringValue.of("ping" + i));
+ assertEquals("re:ping" + i, replies.poll(10, TimeUnit.SECONDS));
+ }
+ requests.onCompleted();
+ done.get(10, TimeUnit.SECONDS);
+ assertTrue(replies.isEmpty());
+ }
+
+ /**
+ * Request bytes the application didn't ask for stay within the queue budget plus HTTP/2 flow
+ * control windows, so a client that respects readiness stops sending.
+ */
+ @Test
+ void unrequestedMessagesPushBackOnTheClient() throws Exception {
+ adapter.setMaxQueuedBytes(32768);
+ CountDownLatch release = new CountDownLatch(1);
+ AtomicInteger received = new AtomicInteger();
+ start(
+ ServerServiceDefinition.builder("review.Server")
+ .addMethod(
+ UPLOAD,
+ ServerCalls.asyncClientStreamingCall(
+ out -> {
+ var observer =
+ (ServerCallStreamObserver) out;
+ observer.disableAutoRequest();
+ Thread.startVirtualThread(
+ () -> {
+ try {
+ release.await();
+ } catch (InterruptedException _) {
+ Thread.currentThread().interrupt();
+ return;
+ }
+ observer.request(Integer.MAX_VALUE);
+ });
+ return new io.grpc.stub.StreamObserver() {
+ @Override
+ public void onNext(StringValue value) {
+ received.incrementAndGet();
+ }
+
+ @Override
+ public void onError(Throwable cause) {
+ /* Not used by this test. */
+ }
+
+ @Override
+ public void onCompleted() {
+ out.onNext(
+ StringValue.of(
+ Integer.toString(
+ received.get())));
+ out.onCompleted();
+ }
+ };
+ }))
+ .build());
+ int total = 1024; // 4 MiB of 4 KiB messages
+ AtomicInteger sent = new AtomicInteger();
+ var result = new java.util.concurrent.CompletableFuture();
+ ClientCall call =
+ channel.newCall(
+ UPLOAD, CallOptions.DEFAULT.withDeadlineAfter(30, TimeUnit.SECONDS));
+ call.start(
+ new ClientCall.Listener<>() {
+ @Override
+ public void onMessage(StringValue message) {
+ result.complete(message.getValue());
+ }
+
+ @Override
+ public void onReady() {
+ while (call.isReady() && sent.get() < total) {
+ call.sendMessage(StringValue.of("x".repeat(4096)));
+ if (sent.incrementAndGet() == total) {
+ call.halfClose();
+ }
+ }
+ }
+
+ @Override
+ public void onClose(Status status, Metadata trailers) {
+ if (!status.isOk()) {
+ result.completeExceptionally(status.asRuntimeException());
+ }
+ }
+ },
+ new Metadata());
+ call.request(1);
+ Thread.sleep(1000);
+ int stalled = sent.get();
+ assertTrue(stalled < total / 2, "sent " + stalled + " of " + total + " messages");
+ assertEquals(0, received.get());
+ release.countDown();
+ assertEquals(Integer.toString(total), result.get(30, TimeUnit.SECONDS));
+ }
+
+ /** A requested message larger than the queue budget is consumed as it arrives. */
+ @Test
+ void messagesLargerThanTheQueueBudgetStreamThrough() throws Exception {
+ adapter.setMaxQueuedBytes(16384);
+ start(echo());
+ String large = "x".repeat(512 * 1024);
+ assertEquals(large, call(large));
+ }
+
+ @Test
+ void cancellingAStreamingCallReleasesItsSlot() throws Exception {
+ adapter.setMaxConcurrentCalls(1);
+ CountDownLatch cancelled = new CountDownLatch(1);
+ CountDownLatch receiving = new CountDownLatch(1);
+ start(
+ ServerServiceDefinition.builder("review.Server")
+ .addMethod(
+ CHAT,
+ ServerCalls.asyncBidiStreamingCall(
+ out -> {
+ ((ServerCallStreamObserver) out)
+ .setOnCancelHandler(cancelled::countDown);
+ return new io.grpc.stub.StreamObserver() {
+ @Override
+ public void onNext(StringValue value) {
+ receiving.countDown();
+ }
+
+ @Override
+ public void onError(Throwable cause) {
+ /* Not used by this test. */
+ }
+
+ @Override
+ public void onCompleted() {
+ out.onCompleted();
+ }
+ };
+ }))
+ .addMethod(
+ ECHO,
+ ServerCalls.asyncUnaryCall(
+ (StringValue request,
+ io.grpc.stub.StreamObserver out) -> {
+ out.onNext(request);
+ out.onCompleted();
+ }))
+ .build());
+ ClientCall call =
+ channel.newCall(CHAT, CallOptions.DEFAULT.withDeadlineAfter(15, TimeUnit.SECONDS));
+ call.start(new ClientCall.Listener<>() {}, new Metadata());
+ call.sendMessage(StringValue.of("open"));
+ assertTrue(receiving.await(10, TimeUnit.SECONDS));
+ call.cancel("client gave up", null);
+ assertTrue(cancelled.await(10, TimeUnit.SECONDS));
+ assertEquals("next", call("next"));
+ }
+
+ private static io.grpc.stub.StreamObserver observer(
+ java.util.concurrent.CompletableFuture result) {
+ return new io.grpc.stub.StreamObserver<>() {
+ @Override
+ public void onNext(StringValue value) {
+ result.complete(value.getValue());
+ }
+
+ @Override
+ public void onError(Throwable cause) {
+ result.completeExceptionally(cause);
+ }
+
+ @Override
+ public void onCompleted() {
+ /* Not used by this test. */
+ }
+ };
+ }
+
private static byte[] frame(String value) throws IOException {
var bytes = new ByteArrayOutputStream();
var data = new DataOutputStream(bytes);
@@ -834,9 +1108,8 @@ public int read() throws IOException {
() -> {});
Response response = new Response(request);
adapter.handle(request, response);
- assertEquals(
- String.valueOf(Status.Code.INVALID_ARGUMENT.value()),
- trailersAfterWriting(response));
+ // A request stream that breaks while read is a peer reset, as grpc-netty reports it.
+ assertEquals(String.valueOf(Status.Code.CANCELLED.value()), trailersAfterWriting(response));
}
@Test
@@ -862,7 +1135,9 @@ public int read() throws IOException {
try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
executor.submit(() -> adapter.handle(request, response)).get(10, TimeUnit.SECONDS);
}
- assertEquals(0, reset.getCount());
+ // The request is read apart from the handler, which returns once the call ends; the read
+ // still stuck after that is reset shortly after.
+ assertTrue(reset.await(10, TimeUnit.SECONDS));
assertEquals(
String.valueOf(Status.Code.DEADLINE_EXCEEDED.value()),
trailersAfterWriting(response));
@@ -918,6 +1193,178 @@ public void write(int value)
}
}
+ /** The end of the request, read while a response write is blocked, doesn't hide that write. */
+ @Test
+ void requestEndDoesNotHideABlockedResponseWrite() throws Exception {
+ adapter.getServices()
+ .add(
+ ServerServiceDefinition.builder("review.Server")
+ .addMethod(
+ CHAT,
+ ServerCalls.asyncBidiStreamingCall(
+ out ->
+ new io.grpc.stub.StreamObserver<
+ StringValue>() {
+ @Override
+ public void onNext(StringValue value) {
+ out.onNext(StringValue.of("first"));
+ }
+
+ @Override
+ public void onError(Throwable cause) {
+ /* Not used by this test. */
+ }
+
+ @Override
+ public void onCompleted() {
+ /* Not used by this test. */
+ }
+ }))
+ .build());
+ adapter.start();
+ CountDownLatch reset = new CountDownLatch(1);
+ CountDownLatch writing = new CountDownLatch(1);
+ byte[] frame = frame("watch");
+ InputStream body =
+ new InputStream() {
+ private boolean sent;
+
+ @Override
+ public int read() throws IOException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public int read(byte[] buffer, int offset, int length) throws IOException {
+ if (!sent) {
+ sent = true;
+ System.arraycopy(frame, 0, buffer, offset, frame.length);
+ return frame.length;
+ }
+ // The request ends only once the response write is blocked.
+ try {
+ writing.await();
+ } catch (InterruptedException _) {
+ Thread.currentThread().interrupt();
+ }
+ return -1;
+ }
+ };
+ Request request = grpcRequest(CHAT, body, reset::countDown);
+ Response response = new Response(request);
+ adapter.handle(request, response);
+ try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
+ var writer =
+ executor.submit(
+ () -> {
+ response.getEntity()
+ .write(
+ new OutputStream() {
+ @Override
+ public void write(int value)
+ throws IOException {
+ writing.countDown();
+ try {
+ // Bounded, so that a missing reset
+ // fails
+ // the test instead of hanging it.
+ reset.await(30, TimeUnit.SECONDS);
+ } catch (InterruptedException _) {
+ Thread.currentThread().interrupt();
+ }
+ throw new IOException("stream reset");
+ }
+ });
+ return null;
+ });
+ assertTrue(writing.await(10, TimeUnit.SECONDS));
+ // Let the reader see the end of the request, then end the call.
+ Thread.sleep(100);
+ adapter.stop();
+ assertTrue(reset.await(10, TimeUnit.SECONDS));
+ assertThrows(
+ java.util.concurrent.ExecutionException.class,
+ () -> writer.get(10, TimeUnit.SECONDS));
+ }
+ }
+
+ /**
+ * A bidirectional service may finish before the client half-closes; the server then closes the
+ * request body, which must not turn the completed call into a cancelled one.
+ */
+ @Test
+ void earlyServerCompletionIsNotACancellation() throws Exception {
+ CountDownLatch completed = new CountDownLatch(1);
+ CountDownLatch cancelled = new CountDownLatch(1);
+ adapter.getInterceptors()
+ .add(
+ new io.grpc.ServerInterceptor() {
+ @Override
+ public io.grpc.ServerCall.Listener interceptCall(
+ io.grpc.ServerCall call,
+ Metadata headers,
+ io.grpc.ServerCallHandler next) {
+ return new io.grpc.ForwardingServerCallListener
+ .SimpleForwardingServerCallListener<>(
+ next.startCall(call, headers)) {
+ @Override
+ public void onComplete() {
+ completed.countDown();
+ super.onComplete();
+ }
+
+ @Override
+ public void onCancel() {
+ cancelled.countDown();
+ super.onCancel();
+ }
+ };
+ }
+ });
+ start(
+ ServerServiceDefinition.builder("review.Server")
+ .addMethod(
+ CHAT,
+ ServerCalls.asyncBidiStreamingCall(
+ out ->
+ new io.grpc.stub.StreamObserver() {
+ @Override
+ public void onNext(StringValue value) {
+ out.onNext(value);
+ out.onCompleted();
+ }
+
+ @Override
+ public void onError(Throwable cause) {
+ /* Not used by this test. */
+ }
+
+ @Override
+ public void onCompleted() {
+ /* Not used by this test. */
+ }
+ }))
+ .build());
+ var status = new java.util.concurrent.CompletableFuture();
+ // The first call on a fresh channel also waits for the HTTP/2 connection setup.
+ ClientCall call =
+ channel.newCall(CHAT, CallOptions.DEFAULT.withDeadlineAfter(30, TimeUnit.SECONDS));
+ call.start(
+ new ClientCall.Listener<>() {
+ @Override
+ public void onClose(Status value, Metadata trailers) {
+ status.complete(value);
+ }
+ },
+ new Metadata());
+ call.request(1);
+ // The client never half-closes.
+ call.sendMessage(StringValue.of("only"));
+ assertEquals(Status.Code.OK, status.get(30, TimeUnit.SECONDS).getCode());
+ assertTrue(completed.await(10, TimeUnit.SECONDS));
+ assertFalse(cancelled.await(200, TimeUnit.MILLISECONDS));
+ }
+
@Test
void callbacksMayUseTheAdapterWhileItStops() throws Exception {
CountDownLatch entered = new CountDownLatch(1);
diff --git a/readme.md b/readme.md
index 436ae556d6..6e0c91d525 100644
--- a/readme.md
+++ b/readme.md
@@ -28,11 +28,11 @@ Restlet 3.0 lets the same application and the same connectors serve APIs in seve
| Style | Extension | Status |
|---|---|---|
| [GraphQL](https://graphql.org/) | `org.restlet.ext.graphql` | Experimental: server and client on graphql-java, with subscriptions over SSE and WebSocket |
-| [gRPC](https://grpc.io/) | `org.restlet.ext.grpc` | Experimental: unary and server-streaming services, and unary client, over HTTP/2 |
+| [gRPC](https://grpc.io/) | `org.restlet.ext.grpc` | Experimental: services of every streaming type, and unary client, over HTTP/2 |
| [MCP](https://modelcontextprotocol.io/) | `org.restlet.ext.mcp` | Experimental: tools, resources and prompts for AI agents on the official MCP Java SDK, over stateless Streamable HTTP and STDIO |
| [Smithy](https://smithy.io/) | `org.restlet.ext.smithy` | smithy-java clients on Restlet HTTP/1.1 and HTTP/2 connectors, discovered or set explicitly; experimental server hosting with the `org.restlet.ext.smithy.server` artifact, on the class path |
-Upcoming enhancements include client and bidirectional streaming for gRPC, and MCP sessions and resumable streams. Feedback is welcome in [Discussions](https://github.com/restlet/restlet-framework-java/discussions).
+Upcoming enhancements include streaming in the gRPC client, and MCP sessions and resumable streams. Feedback is welcome in [Discussions](https://github.com/restlet/restlet-framework-java/discussions).
## Learn more