Host client-streaming and bidirectional gRPC methods

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,

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.
  *
- * <p>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.
+ * <p>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.
  *
  * <p>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());
                     }
                 }
             }

org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletServerHost.java

@@ -323,22 +323,35 @@ private final class Call implements ServerTransport {
 
         private final InternalLogId id = InternalLogId.allocate(Call.class, "Restlet logical call");
         private final ArrayDeque<byte[]> 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();
 
         /** 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;
         private volatile ScheduledFuture<?> deadline;
         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) {