diff --git a/changes.md b/changes.md index 28fa95112c..227390c6f4 100644 --- a/changes.md +++ b/changes.md @@ -107,14 +107,18 @@ Changes log leaves the borrowed listener and supplied callback executor running and reports timeout 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. + - Added an experimental SDK-native gRPC client on a borrowed Restlet HTTP/2 connector, + for unary, server-streaming, client-streaming and bidirectional calls, with explicit + plaintext opt-in, bounded messages and call admission. Each call is one HTTP/2 exchange + whose request body is written while its response is read; request messages are queued + within a per-call budget that the SDK's readiness follows, and response messages are read + as the application requests them, within the same budget. Regression tests cover authority mapping, cancellation before and after headers, disconnect recovery, malformed responses and shared connector ownership. Empty responses retain SDK missing-status diagnostics instead of failing on a null stream. Shutdown arriving before SDK listener installation is replayed once; interrupted shutdown still closes owned transport resources and preserves the interrupt. - TLS, streaming methods and physical gRPC connection controls remain unsupported. + TLS and physical gRPC connection controls remain unsupported. - Added the Smithy extension, with a smithy-java 1.8.0 client transport on Restlet HTTP client connectors. Generated and dynamic SDK clients find it through service loading, before the JDK transport, or are given one borrowing an existing client. Requests are diff --git a/org.restlet.ext.grpc/pom.xml b/org.restlet.ext.grpc/pom.xml index 61a62faf25..909df28fe4 100644 --- a/org.restlet.ext.grpc/pom.xml +++ b/org.restlet.ext.grpc/pom.xml @@ -76,5 +76,11 @@ ${lib-junit-version} test + + org.junit.jupiter + junit-jupiter-params + ${lib-junit-version} + test + diff --git a/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/GrpcClientAdapter.java b/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/GrpcClientAdapter.java index 48950242a0..d23dc1a3dd 100644 --- a/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/GrpcClientAdapter.java +++ b/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/GrpcClientAdapter.java @@ -28,10 +28,14 @@ /** * Single-use native gRPC channel on a borrowed Restlet HTTP/2 client. * - *

This initial profile is unary and plaintext only; applications must explicitly opt in with - * {@link #setPlaintext(boolean)}. TLS/address separation, proxy CONNECT, streaming, retry, - * keepalive and physical-connection controls are not implemented. No SDK network transport is - * selected. The host starts and stops the borrowed connector. + *

Supports unary, server-streaming, client-streaming and bidirectional calls: each call is one + * HTTP/2 exchange whose request body is written while its response is read, so messages flow both + * ways at once. Request messages are queued within {@link #setMaxQueuedBytes}, and the SDK's + * readiness follows their writing; response messages are read as the application requests them, + * within the same budget. This profile is plaintext only; applications must explicitly opt in with + * {@link #setPlaintext(boolean)}. TLS/address separation, proxy CONNECT, retry, keepalive and + * physical-connection controls are not implemented. No SDK network transport is selected. The host + * starts and stops the borrowed connector. */ public final class GrpcClientAdapter implements AutoCloseable { private final Context context; @@ -40,6 +44,7 @@ public final class GrpcClientAdapter implements AutoCloseable { private String target; private boolean plaintext; private int maxMessageBytes = 1024 * 1024; + private int maxQueuedBytes = 256 * 1024; private int maxConcurrentCalls = 128; private Duration maxCallDuration = Duration.ofSeconds(30); private Duration shutdownTimeout = Duration.ofSeconds(5); @@ -100,6 +105,25 @@ public synchronized int getMaxMessageBytes() { return maxMessageBytes; } + /** + * Sets the per-call queue budget, for each direction: request bytes waiting for the network, + * and response bytes read but not yet consumed by the messages the application requested. A + * single message may exceed it. + * + * @param value At least 16384 bytes. + */ + public synchronized void setMaxQueuedBytes(int value) { + requireConfigurable(); + if (value < 16384) { + throw new IllegalArgumentException("Queue budget must be at least 16384 bytes"); + } + maxQueuedBytes = value; + } + + public synchronized int getMaxQueuedBytes() { + return maxQueuedBytes; + } + public synchronized void setMaxConcurrentCalls(int value) { requireConfigurable(); if (value <= 0) { @@ -191,7 +215,11 @@ public synchronized void start() { configured = true; var factory = new RestletTransportFactory( - client, maxMessageBytes, maxConcurrentCalls, shutdownTimeout); + client, + maxMessageBytes, + maxQueuedBytes, + maxConcurrentCalls, + shutdownTimeout); try { var builder = new ManagedChannelImplBuilder(target, () -> factory, () -> 80) diff --git a/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletClientStream.java b/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletClientStream.java index 5b6e6c37bd..8f8c1fc741 100644 --- a/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletClientStream.java +++ b/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletClientStream.java @@ -24,10 +24,13 @@ import io.grpc.internal.TransportFrameUtil; import io.grpc.internal.TransportTracer; import io.grpc.internal.WritableBuffer; -import java.io.ByteArrayOutputStream; import java.io.IOException; +import java.io.InputStream; +import java.io.InterruptedIOException; +import java.io.OutputStream; import java.net.InetSocketAddress; import java.nio.charset.StandardCharsets; +import java.util.ArrayDeque; import java.util.ArrayList; import java.util.Arrays; import java.util.concurrent.ExecutorService; @@ -40,28 +43,71 @@ import org.restlet.data.MediaType; import org.restlet.data.Method; import org.restlet.data.Reference; -import org.restlet.representation.ByteArrayRepresentation; +import org.restlet.representation.OutputRepresentation; +import org.restlet.representation.Representation; import org.restlet.util.Series; -/** SDK-framed, bounded unary exchange. No gRPC frame/status codec is implemented here. */ +/** + * One gRPC call over one HTTP/2 exchange of a borrowed Restlet client, streaming in both + * directions. The SDK frames and deframes messages; this stream only moves their bytes. + * + *

The exchange starts with the request headers. The request body is fed frame by frame from a + * bounded queue, written by the client's background writer, and ends at the half-close; written + * bytes are reported to the SDK, which derives its readiness from them. The response body is read + * as it arrives, while the request is still written: bytes handed to the SDK but not yet consumed + * by its deframer, which only reads the messages the application requested, stay within the queue + * budget too, so a slow consumer stops the reads, and HTTP/2 flow control stops the server. + */ @SuppressWarnings("null") final class RestletClientStream extends AbstractClientStream { + private static final MediaType GRPC = MediaType.valueOf("application/grpc"); + private final SerializingExecutor serial = new SerializingExecutor(Runnable::run); private final State state; private final Client client; private final String target; private final String authority; - private final int wireLimit; + private final int limit; + private final int queueBytes; private final ExecutorService io; private final Consumer finished; - private final ByteArrayOutputStream outbound = new ByteArrayOutputStream(); private final StatsTraceContext stats; + + /** Request frames waiting for the network; also guards the outbound fields below. */ + private final ArrayDeque outbound = new ArrayDeque<>(); + + /** Guards {@link #unread}, and wakes the response reader. */ + private final Object inbound = new Object(); + private Metadata headers; private volatile Request request; private volatile Thread worker; private volatile Deadline deadline; private volatile boolean done; - private boolean sent; + private volatile boolean failed; + + /** Whether the peer reset the stream or the connection failed. */ + private volatile boolean peerGone; + + private boolean started; + + /** Request bytes queued; guarded by {@link #outbound}. */ + private int queued; + + /** Bytes of the latest request message prefix seen, up to 5; guarded by {@link #outbound}. */ + private int prefix; + + /** Message length read from the prefix so far; guarded by {@link #outbound}. */ + private long declared; + + /** Bytes of the latest request message still to come; guarded by {@link #outbound}. */ + private long remaining; + + /** Whether the SDK half-closed the request; guarded by {@link #outbound}. */ + private boolean halfClosed; + + /** Response bytes handed to the SDK but not consumed yet; guarded by {@link #inbound}. */ + private long unread; RestletClientStream( Client client, @@ -72,6 +118,7 @@ final class RestletClientStream extends AbstractClientStream { CallOptions options, ClientStreamTracer[] tracers, int limit, + int queueBytes, ExecutorService io, Consumer finished) { this( @@ -82,6 +129,7 @@ final class RestletClientStream extends AbstractClientStream { headers, options, limit, + queueBytes, io, finished, StatsTraceContext.newClientContext(tracers, Attributes.EMPTY, headers), @@ -96,6 +144,7 @@ private RestletClientStream( Metadata headers, CallOptions options, int limit, + int queueBytes, ExecutorService io, Consumer finished, StatsTraceContext stats, @@ -114,11 +163,13 @@ private RestletClientStream( ip = "[" + ip + "]"; } target = "http://" + ip + ":" + address.getPort() + "/" + method.getFullMethodName(); - wireLimit = limit + 5; + this.limit = limit; + this.queueBytes = queueBytes; this.io = io; this.finished = finished; this.stats = stats; state = new State(limit, stats, tracer, options); + setOnReadyThreshold(Math.max(1, queueBytes / 2)); } @Override @@ -149,39 +200,12 @@ protected Sink abstractClientStreamSink() { return new Sink() { @Override public void writeHeaders(Metadata metadata, byte[] payload) { - headers = metadata; - stats.clientOutboundHeaders(); - serial.execute(state::allocated); + begin(metadata); } @Override public void writeFrame(WritableBuffer frame, boolean end, boolean flush, int messages) { - Status failure = null; - synchronized (outbound) { - if (!done && frame instanceof Buffer buffer) { - if (buffer.readableBytes() > wireLimit - outbound.size()) { - failure = - Status.RESOURCE_EXHAUSTED.withDescription( - "Encoded unary request exceeds byte budget"); - } else { - outbound.write(buffer.data, 0, buffer.size); - } - } - if (frame != null) { - frame.release(); - } - if (failure == null && end && !done && !sent) { - sent = true; - try { - io.execute(RestletClientStream.this::exchange); - } catch (RejectedExecutionException e) { - failure = Status.UNAVAILABLE.withCause(e); - } - } - } - if (failure != null) { - fail(failure); - } + send(frame, end); } @Override @@ -191,7 +215,101 @@ public void cancel(Status status) { }; } + /** Starts the HTTP/2 exchange with the request headers. */ + private void begin(Metadata metadata) { + headers = metadata; + stats.clientOutboundHeaders(); + serial.execute(state::allocated); + Status failure = null; + synchronized (outbound) { + if (!done && !started) { + started = true; + try { + io.execute(this::exchange); + } catch (RejectedExecutionException e) { + failure = Status.UNAVAILABLE.withCause(e); + } + } + } + if (failure != null) { + fail(failure); + } + } + + /** Queues a request frame for the writer, and records the half-close. */ + private void send(WritableBuffer frame, boolean end) { + Status failure = null; + synchronized (outbound) { + if (frame instanceof Buffer buffer && !done && !halfClosed) { + failure = enqueue(buffer); + } + if (frame != null) { + frame.release(); + } + if (end) { + halfClosed = true; + } + outbound.notifyAll(); + } + if (failure != null) { + fail(failure); + } + } + + /** + * Copies a frame into the queue, within its budget; must be called holding {@link #outbound}. A + * message may only start while the queue is within its budget, as it is for a sender that + * respects readiness; once started, it may grow up to the message-size limit, whatever the tail + * of the previous message still queued. The queue so stays within its budget plus one message. + * Messages are delimited by their gRPC prefixes, since frames do not mark where they start: a + * message of unknown length is framed before it is counted. + * + * @return The failure if the budget is exceeded, null otherwise. + */ + private Status enqueue(Buffer buffer) { + int size = buffer.size; + int offset = 0; + while (offset < size) { + if (remaining > 0) { + long skipped = Math.min(remaining, (long) size - offset); + remaining -= skipped; + offset += (int) skipped; + } else if ((prefix == 0 && queued + offset > queueBytes) + || !prefixed(buffer.data[offset])) { + return Status.RESOURCE_EXHAUSTED.withDescription( + "Request queue byte budget exceeded"); + } else { + offset++; + } + } + if (size > 0) { + outbound.addLast(Arrays.copyOf(buffer.data, size)); + queued += size; + onSendingBytes(size); + } + return null; + } + + /** + * Reads a byte of a message prefix: a compression flag, then a 4-byte big-endian length. + * + * @return False if the message exceeds the size limit. + */ + private boolean prefixed(byte value) { + if (prefix > 0) { + declared = (declared << 8) | (value & 0xFF); + } + if (++prefix < 5) { + return true; + } + remaining = declared; + prefix = 0; + declared = 0; + return remaining <= limit; + } + private void fail(Status status) { + failed = true; Deadline current = deadline; Status reported = (status.getCode() == Status.Code.UNAVAILABLE @@ -210,6 +328,7 @@ private void fail(Status status) { state.fail(reported); } }); + wake(); Request active = request; if (active != null) { active.abort(); @@ -220,6 +339,16 @@ private void fail(Status status) { } } + /** Wakes the request writer and the response reader, once the call failed or ended. */ + private void wake() { + synchronized (outbound) { + outbound.notifyAll(); + } + synchronized (inbound) { + inbound.notifyAll(); + } + } + private void exchange() { worker = Thread.currentThread(); Response response = null; @@ -227,86 +356,26 @@ private void exchange() { if (done) { return; } - byte[] body; - synchronized (outbound) { - body = outbound.toByteArray(); - outbound.reset(); - } - Request call = - new Request( - Method.POST, - target, - new ByteArrayRepresentation( - body, MediaType.valueOf("application/grpc"))); - // Resource origin selects the resolved peer; base reference selects HTTP authority. - call.getResourceRef().setBaseRef(new Reference("http://" + authority)); - call.getHeaders().set("te", "trailers"); - byte[][] metadata = TransportFrameUtil.toHttp2Headers(headers); - for (int i = 0; i < metadata.length; i += 2) { - String key = new String(metadata[i], StandardCharsets.US_ASCII); - if (!key.startsWith(":") - && !"content-type".equals(key) - && !"user-agent".equals(key) - && !"te".equals(key)) { - call.getHeaders() - .add(key, new String(metadata[i + 1], StandardCharsets.US_ASCII)); - } - } - call.getClientInfo().setAgent("restlet-grpc/1.85.1"); + Request call = newRequest(); request = call; - if (done || Thread.currentThread().isInterrupted()) { + if (done || failed || Thread.currentThread().isInterrupted()) { call.abort(); return; } + // A disconnect only wakes the reader and the writer: a peer that reset the stream + // after its whole response, as a server finishing before the half-close may, leaves + // that response to be read; one that did not fails the head or the body read. call.getDisconnectSignal() .thenAccept( - cause -> - fail( - Status.UNAVAILABLE - .withDescription("gRPC peer disconnected") - .withCause(cause))); + cause -> { + peerGone = true; + wake(); + }); + // Returns at the response head; the request body is still written meanwhile. response = client.handle(call); - if (done) { - return; - } - if (response.getProtocol() == null - || !"2.0".equals(response.getProtocol().getVersion())) { - fail( - Status.UNAVAILABLE - .withDescription("gRPC requires an HTTP/2 response") - .withCause(response.getStatus().getThrowable())); - return; - } - Metadata initial = metadata(response.getHeaders()); - initial.put( - InternalMetadata.keyOf(":status", Metadata.ASCII_STRING_MARSHALLER), - Integer.toString(response.getStatus().getCode())); - if (response.getEntity() != null && response.getEntity().getMediaType() != null) { - initial.put( - Metadata.Key.of("content-type", Metadata.ASCII_STRING_MARSHALLER), - response.getEntity().getMediaType().toString()); - } - if (response.getHeaders().getFirstValue("grpc-status", true) != null) { - serial.execute(() -> state.trailers(initial)); - return; - } - serial.execute(() -> state.headers(initial)); - if (response.getEntity() != null) { - try (var input = response.getEntity().getStream()) { - if (input != null) { - byte[] bytes = input.readNBytes(wireLimit + 1); - if (bytes.length > wireLimit) { - fail( - Status.RESOURCE_EXHAUSTED.withDescription( - "Encoded unary response exceeds byte budget")); - return; - } - serial.execute(() -> state.data(bytes)); - } - } + if (!done) { + receive(response); } - Metadata trailers = metadata(response.getTrailers()); - serial.execute(() -> state.trailers(trailers)); } catch (IOException | RuntimeException e) { fail(Status.UNAVAILABLE.withDescription("Restlet gRPC exchange failed").withCause(e)); } finally { @@ -317,6 +386,142 @@ private void exchange() { } } + /** Returns the HTTP/2 request of the call, with the gRPC metadata as headers. */ + private Request newRequest() { + Request call = new Request(Method.POST, target, requestEntity()); + // Resource origin selects the resolved peer; base reference selects HTTP authority. + call.getResourceRef().setBaseRef(new Reference("http://" + authority)); + call.getHeaders().set("te", "trailers"); + byte[][] metadata = TransportFrameUtil.toHttp2Headers(headers); + for (int i = 0; i < metadata.length; i += 2) { + String key = new String(metadata[i], StandardCharsets.US_ASCII); + if (!key.startsWith(":") + && !"content-type".equals(key) + && !"user-agent".equals(key) + && !"te".equals(key)) { + call.getHeaders().add(key, new String(metadata[i + 1], StandardCharsets.US_ASCII)); + } + } + call.getClientInfo().setAgent("restlet-grpc/1.85.1"); + return call; + } + + /** Hands the response head, body and trailers to the SDK. */ + private void receive(Response response) throws IOException { + if (response.getProtocol() == null || !"2.0".equals(response.getProtocol().getVersion())) { + fail( + Status.UNAVAILABLE + .withDescription("gRPC requires an HTTP/2 response") + .withCause(response.getStatus().getThrowable())); + return; + } + Metadata initial = metadata(response.getHeaders()); + initial.put( + InternalMetadata.keyOf(":status", Metadata.ASCII_STRING_MARSHALLER), + Integer.toString(response.getStatus().getCode())); + if (response.getEntity() != null && response.getEntity().getMediaType() != null) { + initial.put( + Metadata.Key.of("content-type", Metadata.ASCII_STRING_MARSHALLER), + response.getEntity().getMediaType().toString()); + } + if (response.getHeaders().getFirstValue("grpc-status", true) != null) { + serial.execute(() -> state.trailers(initial)); + return; + } + serial.execute(() -> state.headers(initial)); + if (response.getEntity() != null) { + try (var input = response.getEntity().getStream()) { + if (input != null && !readAll(input)) { + return; + } + } + } + Metadata trailers = metadata(response.getTrailers()); + serial.execute(() -> state.trailers(trailers)); + } + + /** + * Hands the response body to the SDK as it arrives, within the queue budget. + * + * @return False if the call ended before the end of the body. + */ + private boolean readAll(InputStream input) throws IOException { + byte[] chunk = new byte[16384]; + for (; ; ) { + int room; + synchronized (inbound) { + try { + // Once the peer is gone, what the connector received is read regardless. + while (unread >= queueBytes && !done && !failed && !peerGone) { + inbound.wait(); + } + } catch (InterruptedException _) { + Thread.currentThread().interrupt(); + throw new InterruptedIOException("Interrupted reading the gRPC response"); + } + if (done || failed) { + return false; + } + // A live peer is read within the budget; once it is gone, what the connector + // buffered is drained in whole chunks. + room = peerGone ? chunk.length : (int) Math.min(chunk.length, queueBytes - unread); + } + int count = input.read(chunk, 0, room); + if (count < 0) { + return true; + } + if (count > 0) { + synchronized (inbound) { + unread += count; + } + byte[] bytes = Arrays.copyOf(chunk, count); + serial.execute(() -> state.data(bytes)); + } + } + } + + /** Returns the request entity, written by the client's background writer from the queue. */ + private Representation requestEntity() { + return new OutputRepresentation(GRPC, Representation.UNKNOWN_SIZE) { + @Override + public void write(OutputStream output) throws IOException { + for (byte[] data = nextFrame(); data != null; data = nextFrame()) { + output.write(data); + output.flush(); + int sent = data.length; + serial.execute(() -> state.sent(sent)); + } + } + }; + } + + /** + * Waits for the next request frame to write. + * + * @return The frame, or null once the request ends normally: half-closed, or the call ended. + * @throws IOException If the call failed or the peer is gone, so the request is not complete. + */ + private byte[] nextFrame() throws IOException { + synchronized (outbound) { + try { + while (outbound.isEmpty() && !halfClosed && !done && !failed && !peerGone) { + outbound.wait(); + } + } catch (InterruptedException _) { + Thread.currentThread().interrupt(); + throw new InterruptedIOException("Interrupted writing the gRPC request"); + } + if (failed || peerGone) { + throw new IOException("gRPC request not completed"); + } + byte[] data = outbound.pollFirst(); + if (data != null) { + queued -= data.length; + } + return data; + } + } + private static Metadata metadata(Series

headers) { var values = new ArrayList(); for (Header header : headers) { @@ -331,7 +536,7 @@ private static Metadata metadata(Series
headers) { } private final class State extends Http2ClientStreamTransportState { - private boolean started; + private boolean allocatedOnce; private Status pendingFailure; private State( @@ -340,7 +545,7 @@ private State( } private void allocated() { - started = true; + allocatedOnce = true; if (!done) { if (pendingFailure != null) { transportReportStatus(pendingFailure, true, new Metadata()); @@ -351,7 +556,7 @@ private void allocated() { } private void fail(Status status) { - if (started) { + if (allocatedOnce) { transportReportStatus(status, true, new Metadata()); } else if (pendingFailure == null) { // Shutdown may arrive after stream creation but before SDK listener installation. @@ -377,6 +582,12 @@ private void trailers(Metadata metadata) { } } + private void sent(int count) { + if (!done) { + onSentBytes(count); + } + } + @Override public void runOnTransportThread(Runnable runnable) { serial.execute(runnable); @@ -384,7 +595,11 @@ public void runOnTransportThread(Runnable runnable) { @Override public void bytesRead(int count) { - // Unary wire input is bounded and materialized before SDK deframing. + // The deframer consumed response bytes: free that much of the inbound budget. + synchronized (inbound) { + unread -= count; + inbound.notifyAll(); + } } @Override @@ -405,10 +620,33 @@ public void deframerClosed(boolean partial) { // and a caller that starts its next call right away must not find the slot still taken. if (!done) { done = true; + wake(); + endRequest(); finished.accept(RestletClientStream.this); } super.deframerClosed(partial); } + + /** + * Ends the request once the call ended, as when a server answers before the half-close: its + * queued frames are dropped and the exchange is aborted, which resets an HTTP/2 stream + * whose request is unfinished, even once the entity writer returned, and releases a writer + * blocked by flow control. The connector ignores the abort of a completed exchange, and the + * call's own status is already reported. + */ + private void endRequest() { + boolean begun; + synchronized (outbound) { + begun = started; + outbound.clear(); + queued = 0; + outbound.notifyAll(); + } + Request active = request; + if (begun && active != null) { + active.abort(); + } + } } private static final class Buffer implements WritableBuffer { diff --git a/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletTransportFactory.java b/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletTransportFactory.java index dd28450b53..f60108b2f0 100644 --- a/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletTransportFactory.java +++ b/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletTransportFactory.java @@ -43,6 +43,7 @@ public final class RestletTransportFactory implements ClientTransportFactory { private final Client client; private final int maxBytes; + private final int queueBytes; private final Duration shutdownTimeout; private final Semaphore admission; private final Set transports = new HashSet<>(); @@ -52,9 +53,11 @@ public final class RestletTransportFactory implements ClientTransportFactory { 1, Thread.ofPlatform().daemon().name("restlet-grpc-timer").factory()); private boolean closed; - public RestletTransportFactory(Client client, int maxBytes, int maxCalls, Duration timeout) { + public RestletTransportFactory( + Client client, int maxBytes, int queueBytes, int maxCalls, Duration timeout) { this.client = client; this.maxBytes = maxBytes; + this.queueBytes = queueBytes; admission = new Semaphore(maxCalls); shutdownTimeout = timeout; // Cancelled deadline timers of completed calls must not stay queued until they expire. @@ -161,8 +164,10 @@ public synchronized ClientStream newStream( Status.UNIMPLEMENTED.withDescription( "Per-call authority override is not supported"); } - if (rejected == null && method.getType() != MethodDescriptor.MethodType.UNARY) { - rejected = Status.UNIMPLEMENTED.withDescription("R3 supports unary methods only"); + if (rejected == null && method.getType() == MethodDescriptor.MethodType.UNKNOWN) { + rejected = + Status.UNIMPLEMENTED.withDescription( + "Methods of unknown type are not supported"); } if (rejected == null && options.getCompressor() != null @@ -189,6 +194,7 @@ public synchronized ClientStream newStream( options, tracers, maxBytes, + queueBytes, io, this::finished); streams.add(stream); diff --git a/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcClientAdapterTestCase.java b/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcClientAdapterTestCase.java index 5b19cde07b..da21769dd8 100644 --- a/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcClientAdapterTestCase.java +++ b/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcClientAdapterTestCase.java @@ -373,7 +373,9 @@ public void onClose(Status value, Metadata trailers) { assertEquals(Status.Code.CANCELLED, status.get().getCode()); assertEquals("recovered", call("recovered").getValue()); assertEquals(1, terminals.get()); - assertEquals(0, messages.get()); + // Responses stream as they arrive, so the one message sent before the server waited may + // be delivered before the cancellation, as with grpc-netty; never more. + assertTrue(messages.get() <= 1, "messages: " + messages.get()); } private record Pending(CountDownLatch entered, CountDownLatch cancelled) {} @@ -522,7 +524,7 @@ void interruptedStopStillClosesOwnedChannelAndPreservesBorrowedResources() throw } @Test - void rejectsStreamingAndOutboundLimitBeforeSending() throws Exception { + void rejectsUnknownMethodTypesAndOutboundLimitBeforeSending() throws Exception { adapter.setMaxMessageBytes(128); start(echo()); assertEquals( @@ -531,8 +533,8 @@ void rejectsStreamingAndOutboundLimitBeforeSending() throws Exception { .getStatus() .getCode()); assertEquals("small", call("small").getValue()); - var streamMethod = - ECHO.toBuilder().setType(MethodDescriptor.MethodType.SERVER_STREAMING).build(); + // Streaming methods are supported; only methods of unknown type are refused. + var streamMethod = ECHO.toBuilder().setType(MethodDescriptor.MethodType.UNKNOWN).build(); assertEquals( Status.Code.UNIMPLEMENTED, assertThrows( diff --git a/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcClientStreamingTestCase.java b/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcClientStreamingTestCase.java new file mode 100644 index 0000000000..ab1a1533f4 --- /dev/null +++ b/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcClientStreamingTestCase.java @@ -0,0 +1,812 @@ +/** + * Copyright 2005-2026 Qlik + *

+ * The content of this file is subject to the terms of the Apache 2.0 open + * source license available at https://www.opensource.org/licenses/apache-2.0 + *

+ * Restlet is a registered trademark of QlikTech International AB. + */ +package org.restlet.ext.grpc; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import com.google.protobuf.StringValue; +import io.grpc.CallOptions; +import io.grpc.ClientCall; +import io.grpc.Metadata; +import io.grpc.MethodDescriptor; +import io.grpc.ServerServiceDefinition; +import io.grpc.Status; +import io.grpc.StatusRuntimeException; +import io.grpc.netty.shaded.io.grpc.netty.NettyServerBuilder; +import io.grpc.protobuf.ProtoUtils; +import io.grpc.stub.ClientCalls; +import io.grpc.stub.ServerCallStreamObserver; +import io.grpc.stub.ServerCalls; +import io.grpc.stub.StreamObserver; +import java.io.FilterInputStream; +import java.io.InputStream; +import java.time.Duration; +import java.util.ArrayList; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.Semaphore; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; +import org.restlet.Client; +import org.restlet.Context; +import org.restlet.Server; +import org.restlet.data.Protocol; +import org.restlet.routing.Router; + +/** + * Streaming calls of every type through {@link GrpcClientAdapter}, against grpc-java's own Netty + * server and against a {@link GrpcServerAdapter}, so that both ends may be Restlet. + */ +class GrpcClientStreamingTestCase { + private static final MethodDescriptor ECHO = + MethodDescriptor.newBuilder() + .setType(MethodDescriptor.MethodType.UNARY) + .setFullMethodName("review.Streams/Echo") + .setRequestMarshaller(ProtoUtils.marshaller(StringValue.getDefaultInstance())) + .setResponseMarshaller(ProtoUtils.marshaller(StringValue.getDefaultInstance())) + .build(); + private static final MethodDescriptor WATCH = + ECHO.toBuilder() + .setType(MethodDescriptor.MethodType.SERVER_STREAMING) + .setFullMethodName("review.Streams/Watch") + .build(); + private static final MethodDescriptor UPLOAD = + ECHO.toBuilder() + .setType(MethodDescriptor.MethodType.CLIENT_STREAMING) + .setFullMethodName("review.Streams/Upload") + .build(); + private static final MethodDescriptor CHAT = + ECHO.toBuilder() + .setType(MethodDescriptor.MethodType.BIDI_STREAMING) + .setFullMethodName("review.Streams/Chat") + .build(); + + /** Upload, with request messages of unknown length, which grpc-java frames before counting. */ + private static final MethodDescriptor UNKNOWN_LENGTH_UPLOAD = + UPLOAD.toBuilder( + new MethodDescriptor.Marshaller() { + @Override + public InputStream stream(StringValue value) { + return new FilterInputStream( + ECHO.getRequestMarshaller().stream(value)) {}; + } + + @Override + public StringValue parse(InputStream stream) { + return ECHO.getRequestMarshaller().parse(stream); + } + }, + ECHO.getResponseMarshaller()) + .build(); + + private static final String BLOCK = "x".repeat(4096); + + private final GrpcClientAdapter adapter = new GrpcClientAdapter(); + private Client client; + private io.grpc.Server nettyPeer; + private GrpcServerAdapter restletPeer; + private Server http; + + /** Starts the peer, Restlet or Netty, and the adapter calling it. */ + private void start(boolean restlet, ServerServiceDefinition service) throws Exception { + int port; + if (restlet) { + restletPeer = new GrpcServerAdapter(); + // Above the 5 s default: the demand tests stall on purpose before moving 4 MiB. + restletPeer.setMaxCallDuration(Duration.ofSeconds(30)); + restletPeer.getServices().add(service); + restletPeer.start(); + Router router = new Router(); + restletPeer.attachTo(router); + Context context = new Context(); + context.getParameters().set("http.transport.protocol", "HTTP2"); + http = new Server(context, Protocol.HTTP, 0, router); + http.start(); + port = http.getActualPort(); + } else { + // A fixed flow control window, so that the peer cannot absorb what the client holds. + nettyPeer = + NettyServerBuilder.forPort(0) + .flowControlWindow(65536) + .addService(service) + .build() + .start(); + port = nettyPeer.getPort(); + } + Context context = new Context(); + context.getParameters().set("httpClientTransportMode", "HTTP2"); + context.getParameters().set("followRedirects", "false"); + client = new Client(context, Protocol.HTTP); + client.start(); + adapter.setClient(client); + adapter.setTarget("dns:///localhost:" + port); + adapter.setPlaintext(true); + adapter.start(); + } + + @AfterEach + void cleanup() throws Exception { + try { + adapter.stop(); + } finally { + try { + if (client != null) { + client.stop(); + } + } finally { + if (nettyPeer != null) { + nettyPeer.shutdownNow(); + assertTrue(nettyPeer.awaitTermination(30, TimeUnit.SECONDS)); + } + if (restletPeer != null) { + restletPeer.stop(); + } + if (http != null) { + http.stop(); + } + } + } + } + + private static CallOptions deadline() { + return CallOptions.DEFAULT.withDeadlineAfter(30, TimeUnit.SECONDS); + } + + /** A service with one method of each streaming type. */ + private static ServerServiceDefinition streams() { + return ServerServiceDefinition.builder("review.Streams") + .addMethod( + ECHO, + ServerCalls.asyncUnaryCall( + (StringValue request, StreamObserver out) -> { + out.onNext(request); + out.onCompleted(); + })) + .addMethod( + WATCH, + ServerCalls.asyncServerStreamingCall( + (StringValue request, StreamObserver out) -> { + int count = Integer.parseInt(request.getValue()); + for (int i = 0; i < count; i++) { + out.onNext(StringValue.of(i + ":" + BLOCK)); + } + out.onCompleted(); + })) + .addMethod( + UPLOAD, + ServerCalls.asyncClientStreamingCall( + out -> + new StreamObserver() { + private int messages; + private long characters; + + @Override + public void onNext(StringValue value) { + messages++; + characters += value.getValue().length(); + } + + @Override + public void onError(Throwable cause) { + // The call ended; nothing to answer. + } + + @Override + public void onCompleted() { + out.onNext( + StringValue.of( + messages + ":" + characters)); + out.onCompleted(); + } + })) + .addMethod( + CHAT, + ServerCalls.asyncBidiStreamingCall( + out -> + new StreamObserver() { + @Override + public void onNext(StringValue value) { + out.onNext( + StringValue.of("re:" + value.getValue())); + } + + @Override + public void onError(Throwable cause) { + // The call ended; nothing to answer. + } + + @Override + public void onCompleted() { + out.onCompleted(); + } + })) + .build(); + } + + /** The producer ignores readiness, so the stream stays within the 256 KiB queue budget. */ + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void serverStreamingPreservesOrder(boolean restlet) throws Exception { + start(restlet, streams()); + var responses = + ClientCalls.blockingServerStreamingCall( + adapter.getChannel(), WATCH, deadline(), StringValue.of("50")); + var values = new ArrayList(); + responses.forEachRemaining(value -> values.add(value.getValue())); + assertEquals(50, values.size()); + for (int i = 0; i < 50; i++) { + assertEquals(i + ":" + BLOCK, values.get(i)); + } + } + + /** The sender ignores readiness, so the stream stays within the 256 KiB queue budget. */ + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void clientStreamingSendsEveryMessage(boolean restlet) throws Exception { + start(restlet, streams()); + var result = new CompletableFuture(); + var requests = + ClientCalls.asyncClientStreamingCall( + adapter.getChannel().newCall(UPLOAD, deadline()), single(result)); + for (int i = 0; i < 50; i++) { + requests.onNext(StringValue.of(BLOCK)); + } + requests.onCompleted(); + assertEquals("50:" + (50 * BLOCK.length()), result.get(30, TimeUnit.SECONDS)); + } + + /** Each reply is received before the next request is sent, over one HTTP/2 stream. */ + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void bidirectionalCallsExchangeMessagesBothWays(boolean restlet) throws Exception { + start(restlet, streams()); + var replies = new LinkedBlockingQueue(); + var done = new CompletableFuture(); + var requests = + ClientCalls.asyncBidiStreamingCall( + adapter.getChannel().newCall(CHAT, deadline()), + new 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(30, TimeUnit.SECONDS)); + } + requests.onCompleted(); + done.get(30, TimeUnit.SECONDS); + assertTrue(replies.isEmpty()); + } + + /** A client respecting readiness stops sending while the server requests nothing. */ + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void requestsFollowTheServerDemand(boolean restlet) throws Exception { + adapter.setMaxQueuedBytes(32768); + CountDownLatch release = new CountDownLatch(1); + AtomicInteger received = new AtomicInteger(); + start( + restlet, + ServerServiceDefinition.builder("review.Streams") + .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 StreamObserver() { + @Override + public void onNext(StringValue value) { + received.incrementAndGet(); + } + + @Override + public void onError(Throwable cause) { + // The call ended; nothing to answer. + } + + @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 CompletableFuture(); + ClientCall call = + adapter.getChannel().newCall(UPLOAD, deadline()); + 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(BLOCK)); + 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); + // Let the sender fill every buffer on the way. + int stalled = settled(sent); + 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 server respecting readiness stops sending while the client requests nothing. */ + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void responsesFollowTheClientDemand(boolean restlet) throws Exception { + adapter.setMaxQueuedBytes(32768); + int total = 1024; + AtomicInteger produced = new AtomicInteger(); + start( + restlet, + ServerServiceDefinition.builder("review.Streams") + .addMethod( + WATCH, + ServerCalls.asyncServerStreamingCall( + (StringValue request, StreamObserver out) -> { + var observer = + (ServerCallStreamObserver) out; + observer.setOnReadyHandler( + () -> { + while (observer.isReady() + && produced.get() < total) { + produced.incrementAndGet(); + observer.onNext(StringValue.of(BLOCK)); + } + if (produced.compareAndSet( + total, total + 1)) { + observer.onCompleted(); + } + }); + })) + .build()); + AtomicInteger received = new AtomicInteger(); + var closed = new CompletableFuture(); + ClientCall call = adapter.getChannel().newCall(WATCH, deadline()); + call.start( + new ClientCall.Listener<>() { + @Override + public void onMessage(StringValue message) { + received.incrementAndGet(); + } + + @Override + public void onClose(Status status, Metadata trailers) { + closed.complete(status); + } + }, + new Metadata()); + call.sendMessage(StringValue.of("go")); + call.halfClose(); + // Requests nothing yet: the server fills every buffer on the way, then stops. + int stalled = settled(produced); + assertTrue(stalled < total / 2, "produced " + stalled + " of " + total + " messages"); + assertEquals(0, received.get()); + call.request(Integer.MAX_VALUE); + assertEquals(Status.Code.OK, closed.get(30, TimeUnit.SECONDS).getCode()); + assertEquals(total, received.get()); + } + + /** Messages larger than the queue budget still stream through, in both directions. */ + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void messagesLargerThanTheQueueBudgetStreamThrough(boolean restlet) throws Exception { + adapter.setMaxQueuedBytes(16384); + start(restlet, streams()); + String large = "x".repeat(512 * 1024); + assertEquals( + large, + ClientCalls.blockingUnaryCall( + adapter.getChannel(), ECHO, deadline(), StringValue.of(large)) + .getValue()); + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void cancellingAStreamReachesTheServerAndFreesTheSlot(boolean restlet) throws Exception { + adapter.setMaxConcurrentCalls(1); + CountDownLatch receiving = new CountDownLatch(1); + CountDownLatch cancelled = new CountDownLatch(1); + start( + restlet, + ServerServiceDefinition.builder("review.Streams") + .addMethod( + CHAT, + ServerCalls.asyncBidiStreamingCall( + out -> { + ((ServerCallStreamObserver) out) + .setOnCancelHandler(cancelled::countDown); + return new StreamObserver() { + @Override + public void onNext(StringValue value) { + receiving.countDown(); + } + + @Override + public void onError(Throwable cause) { + // The call ended; nothing to answer. + } + + @Override + public void onCompleted() { + out.onCompleted(); + } + }; + })) + .addMethod( + ECHO, + ServerCalls.asyncUnaryCall( + (StringValue request, StreamObserver out) -> { + out.onNext(request); + out.onCompleted(); + })) + .build()); + var closed = new CompletableFuture(); + ClientCall call = adapter.getChannel().newCall(CHAT, deadline()); + call.start( + new ClientCall.Listener<>() { + @Override + public void onClose(Status status, Metadata trailers) { + closed.complete(status); + } + }, + new Metadata()); + call.sendMessage(StringValue.of("open")); + assertTrue(receiving.await(30, TimeUnit.SECONDS)); + call.cancel("client gave up", null); + assertEquals(Status.Code.CANCELLED, closed.get(30, TimeUnit.SECONDS).getCode()); + assertTrue(cancelled.await(30, TimeUnit.SECONDS)); + assertEquals( + "next", + ClientCalls.blockingUnaryCall( + adapter.getChannel(), ECHO, deadline(), StringValue.of("next")) + .getValue()); + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void anOpenStreamEndsAtItsDeadline(boolean restlet) throws Exception { + start(restlet, streams()); + var requests = new CompletableFuture>(); + var error = new CompletableFuture(); + requests.complete( + ClientCalls.asyncBidiStreamingCall( + adapter.getChannel() + .newCall( + CHAT, + CallOptions.DEFAULT.withDeadlineAfter( + 500, TimeUnit.MILLISECONDS)), + new StreamObserver() { + @Override + public void onNext(StringValue value) { + // Replies are not checked here. + } + + @Override + public void onError(Throwable cause) { + error.complete(cause); + } + + @Override + public void onCompleted() { + error.complete(null); + } + })); + // Never half-closed: only the deadline ends the call. + requests.get().onNext(StringValue.of("open")); + Throwable cause = error.get(30, TimeUnit.SECONDS); + assertTrue(cause instanceof StatusRuntimeException, String.valueOf(cause)); + assertEquals( + Status.Code.DEADLINE_EXCEEDED, + ((StatusRuntimeException) cause).getStatus().getCode()); + } + + /** + * A server may answer before the client half-closes, while the request is still written and + * held by flow control: the client gets the answer and its status, and the unfinished request + * is reset, so the only call slot is free at once. + */ + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void anEarlyAnswerEndsTheCallWhileTheRequestIsStillSent(boolean restlet) throws Exception { + adapter.setMaxConcurrentCalls(1); + start( + restlet, + ServerServiceDefinition.builder("review.Streams") + .addMethod( + UPLOAD, + ServerCalls.asyncClientStreamingCall( + out -> + new StreamObserver() { + private boolean answered; + + @Override + public void onNext(StringValue value) { + if (!answered) { + answered = true; + out.onNext(StringValue.of("enough")); + out.onCompleted(); + } + } + + @Override + public void onError(Throwable cause) { + // The call ended; nothing to answer. + } + + @Override + public void onCompleted() { + // Already answered. + } + })) + .addMethod( + ECHO, + ServerCalls.asyncUnaryCall( + (StringValue request, StreamObserver out) -> { + out.onNext(request); + out.onCompleted(); + })) + .build()); + var reply = new CompletableFuture(); + var closed = new CompletableFuture(); + AtomicInteger sent = new AtomicInteger(); + ClientCall call = + adapter.getChannel().newCall(UPLOAD, deadline()); + call.start( + new ClientCall.Listener<>() { + @Override + public void onMessage(StringValue message) { + reply.complete(message.getValue()); + } + + @Override + public void onReady() { + // Never half-closes: keeps sending while ready, up to 4 MiB. + while (call.isReady() && sent.get() < 1024 && !closed.isDone()) { + call.sendMessage(StringValue.of(BLOCK)); + sent.incrementAndGet(); + } + } + + @Override + public void onClose(Status status, Metadata trailers) { + closed.complete(status); + } + }, + new Metadata()); + call.request(1); + assertEquals(Status.Code.OK, closed.get(30, TimeUnit.SECONDS).getCode()); + assertEquals("enough", reply.get(30, TimeUnit.SECONDS)); + assertEquals( + "next", + ClientCalls.blockingUnaryCall( + adapter.getChannel(), ECHO, deadline(), StringValue.of("next")) + .getValue()); + } + + /** + * With a direct callback executor, readiness may return while the tail of a large message is + * still queued: a sender respecting it may start the next large message. + */ + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void largeMessagesSentOnReadinessAreAccepted(boolean restlet) throws Exception { + assertLargeMessagesSentOnReadinessAreAccepted(restlet, UPLOAD); + } + + /** + * A message of unknown length is framed before grpc-java counts it: it still starts within the + * queue budget, and grows to its size. + */ + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void largeMessagesOfUnknownLengthAreAccepted(boolean restlet) throws Exception { + assertLargeMessagesSentOnReadinessAreAccepted(restlet, UNKNOWN_LENGTH_UPLOAD); + } + + private void assertLargeMessagesSentOnReadinessAreAccepted( + boolean restlet, MethodDescriptor upload) throws Exception { + adapter.setMaxQueuedBytes(16384); + adapter.setExecutor(Runnable::run); + start(restlet, streams()); + String large = "y".repeat(100 * 1024); + var result = new CompletableFuture(); + AtomicInteger sent = new AtomicInteger(); + ClientCall call = + adapter.getChannel().newCall(upload, deadline()); + call.start( + new ClientCall.Listener<>() { + @Override + public void onMessage(StringValue message) { + result.complete(message.getValue()); + } + + @Override + public void onReady() { + while (call.isReady() && sent.get() < 2) { + call.sendMessage(StringValue.of(large)); + if (sent.incrementAndGet() == 2) { + call.halfClose(); + } + } + } + + @Override + public void onClose(Status status, Metadata trailers) { + if (!status.isOk()) { + result.completeExceptionally(status.asRuntimeException()); + } + } + }, + new Metadata()); + call.request(1); + assertEquals("2:" + (2 * large.length()), result.get(30, TimeUnit.SECONDS)); + } + + /** + * A sender ignoring readiness towards a peer that requests nothing fills the request queue: the + * call ends once with RESOURCE_EXHAUSTED, the stream is reset and the only call slot is free. + */ + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void aSenderIgnoringReadinessFailsAtTheQueueBudget(boolean restlet) throws Exception { + adapter.setMaxConcurrentCalls(1); + adapter.setMaxQueuedBytes(16384); + start( + restlet, + ServerServiceDefinition.builder("review.Streams") + .addMethod( + CHAT, + ServerCalls.asyncBidiStreamingCall( + out -> { + // A stalled peer: it never asks for a message. + ((ServerCallStreamObserver) out) + .disableAutoRequest(); + return new StreamObserver() { + @Override + public void onNext(StringValue value) { + // Never requested. + } + + @Override + public void onError(Throwable cause) { + // The call ended; nothing to answer. + } + + @Override + public void onCompleted() { + out.onCompleted(); + } + }; + })) + .addMethod( + ECHO, + ServerCalls.asyncUnaryCall( + (StringValue request, StreamObserver out) -> { + out.onNext(request); + out.onCompleted(); + })) + .build()); + var closed = new CompletableFuture(); + AtomicInteger closes = new AtomicInteger(); + ClientCall call = adapter.getChannel().newCall(CHAT, deadline()); + call.start( + new ClientCall.Listener<>() { + @Override + public void onClose(Status status, Metadata trailers) { + closes.incrementAndGet(); + closed.complete(status); + } + }, + new Metadata()); + // Up to 4 MiB, far beyond what the peer, the connectors and the queue may hold. + for (int i = 0; i < 1024 && !closed.isDone(); i++) { + call.sendMessage(StringValue.of(BLOCK)); + } + Status status = closed.get(30, TimeUnit.SECONDS); + assertEquals(Status.Code.RESOURCE_EXHAUSTED, status.getCode()); + assertEquals("Request queue byte budget exceeded", status.getDescription()); + assertEquals( + "next", + ClientCalls.blockingUnaryCall( + adapter.getChannel(), ECHO, deadline(), StringValue.of("next")) + .getValue()); + assertEquals(1, closes.get()); + } + + /** + * Waits until a counter stops moving for 300 ms, as the buffers on the way are full, or 10 s. + * + * @return The settled value. + */ + private static int settled(AtomicInteger counter) throws InterruptedException { + var never = new Semaphore(0); + long end = System.nanoTime() + TimeUnit.SECONDS.toNanos(10); + int last; + int value = counter.get(); + do { + last = value; + never.tryAcquire(300, TimeUnit.MILLISECONDS); + value = counter.get(); + } while (value != last && System.nanoTime() < end); + return value; + } + + private static StreamObserver single(CompletableFuture result) { + return new StreamObserver<>() { + @Override + public void onNext(StringValue value) { + result.complete(value.getValue()); + } + + @Override + public void onError(Throwable cause) { + result.completeExceptionally(cause); + } + + @Override + public void onCompleted() { + // The single reply already completed the result. + } + }; + } +} 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 da4d3b61b4..67702809c4 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 @@ -509,6 +509,8 @@ public void onCompleted() { @Test void unrequestedMessagesPushBackOnTheClient() throws Exception { adapter.setMaxQueuedBytes(32768); + // Above the 5 s default: the test stalls on purpose before moving 4 MiB. + adapter.setMaxCallDuration(Duration.ofSeconds(30)); CountDownLatch release = new CountDownLatch(1); AtomicInteger received = new AtomicInteger(); start( diff --git a/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/internal/RestletClientStreamTestCase.java b/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/internal/RestletClientStreamTestCase.java index e6b3ba14e4..900778a571 100644 --- a/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/internal/RestletClientStreamTestCase.java +++ b/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/internal/RestletClientStreamTestCase.java @@ -36,7 +36,8 @@ class RestletClientStreamTestCase { @Test void cancelledTimersDoNotStayQueued() { - var factory = new RestletTransportFactory(null, 1024, 1, java.time.Duration.ofSeconds(1)); + var factory = + new RestletTransportFactory(null, 1024, 16384, 1, java.time.Duration.ofSeconds(1)); try { var timer = (ScheduledThreadPoolExecutor) factory.getScheduledExecutorService(); for (int i = 0; i < 100; i++) { @@ -79,6 +80,7 @@ private static RestletClientStream stream(ExecutorService io, AtomicInteger rele CallOptions.DEFAULT, new ClientStreamTracer[0], 128, + 16384, io, ignored -> released.incrementAndGet()); } diff --git a/readme.md b/readme.md index 6e0c91d525..71d3caf222 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: services of every streaming type, and unary client, over HTTP/2 | +| [gRPC](https://grpc.io/) | `org.restlet.ext.grpc` | Experimental: services and clients of every streaming type, 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 streaming in the gRPC client, and MCP sessions and resumable streams. Feedback is welcome in [Discussions](https://github.com/restlet/restlet-framework-java/discussions). +Upcoming enhancements include MCP sessions and resumable streams. Feedback is welcome in [Discussions](https://github.com/restlet/restlet-framework-java/discussions). ## Learn more