From f7fe11a32abf36117543001dab901f50e439b75e Mon Sep 17 00:00:00 2001
From: Jerome Louvel <374450+jlouvel@users.noreply.github.com>
Date: Fri, 23 Oct 2026 20:30:33 -0400
Subject: [PATCH 1/4] Add unary-first SDK-native gRPC client transport on
Restlet
Reuse a borrowed Restlet HTTP/2 connector with grpc-java 1.85.1. Bound unary messages and calls, preserve SDK lifecycle semantics, and verify packaged named-module consumption.
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Signed-off-by: Jerome Louvel <374450+jlouvel@users.noreply.github.com>
---
changes.md | 8 +
org.restlet.ext.grpc/pom.xml | 75 ++
.../src/main/java/module-info.java | 17 +
.../restlet/ext/grpc/GrpcClientAdapter.java | 290 +++++++
.../grpc/internal/GuardedClientStream.java | 128 +++
.../grpc/internal/RestletClientStream.java | 427 ++++++++++
.../internal/RestletTransportFactory.java | 262 ++++++
.../ext/grpc/GrpcClientAdapterTestCase.java | 759 ++++++++++++++++++
.../ext/grpc/GrpcHttpValidationTestCase.java | 283 +++++++
.../restlet/ext/grpc/GrpcModuleTestCase.java | 142 ++++
.../internal/RestletClientStreamTestCase.java | 100 +++
pom.xml | 2 +
12 files changed, 2493 insertions(+)
create mode 100644 org.restlet.ext.grpc/pom.xml
create mode 100644 org.restlet.ext.grpc/src/main/java/module-info.java
create mode 100644 org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/GrpcClientAdapter.java
create mode 100644 org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/GuardedClientStream.java
create mode 100644 org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletClientStream.java
create mode 100644 org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletTransportFactory.java
create mode 100644 org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcClientAdapterTestCase.java
create mode 100644 org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcHttpValidationTestCase.java
create mode 100644 org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcModuleTestCase.java
create mode 100644 org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/internal/RestletClientStreamTestCase.java
diff --git a/changes.md b/changes.md
index c4484e2572..4106e5e0a2 100644
--- a/changes.md
+++ b/changes.md
@@ -81,6 +81,14 @@ Changes log
supporting helper class. Applications with multiple STDIO protocol helpers must
select one explicitly.
- Enhancements
+ - 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,
+ 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.
- Added the supported `org.restlet.util.StreamTransport` seam for protocol-neutral,
bounded synchronous byte I/O, serialized writes, borrowed-stream cancellation and
owned subprocess lifecycle with separate bounded stderr capture. No protocol
diff --git a/org.restlet.ext.grpc/pom.xml b/org.restlet.ext.grpc/pom.xml
new file mode 100644
index 0000000000..8e4f53f1f1
--- /dev/null
+++ b/org.restlet.ext.grpc/pom.xml
@@ -0,0 +1,75 @@
+
+
+ 4.0.0
+
+ org.restlet
+ org.restlet.parent
+ 3.0.0-SNAPSHOT
+
+ org.restlet.ext.grpc
+ Restlet Framework - gRPC extension
+ Native gRPC unary client on a borrowed Restlet HTTP/2 connector.
+
+
+
+ com.google.guava
+ guava
+ 33.6.0-android
+
+
+ com.google.errorprone
+ error_prone_annotations
+ 2.50.0
+
+
+ io.grpc
+ grpc-bom
+ ${lib-grpc-version}
+ pom
+ import
+
+
+
+
+
+ com.google.guava
+ guava
+
+
+ org.restlet
+ org.restlet
+ ${project.version}
+
+
+ io.grpc
+ grpc-core
+
+
+ io.grpc
+ grpc-stub
+ test
+
+
+ io.grpc
+ grpc-protobuf
+ test
+
+
+ io.grpc
+ grpc-netty-shaded
+ test
+
+
+ io.grpc
+ grpc-services
+ test
+
+
+ org.junit.jupiter
+ junit-jupiter-api
+ ${lib-junit-version}
+ test
+
+
+
diff --git a/org.restlet.ext.grpc/src/main/java/module-info.java b/org.restlet.ext.grpc/src/main/java/module-info.java
new file mode 100644
index 0000000000..e5da89d47b
--- /dev/null
+++ b/org.restlet.ext.grpc/src/main/java/module-info.java
@@ -0,0 +1,17 @@
+/**
+ * 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.
+ */
+@SuppressWarnings({"requires-automatic", "requires-transitive-automatic"})
+module org.restlet.ext.grpc {
+ requires transitive org.restlet;
+ requires transitive io.grpc;
+ requires io.grpc.internal;
+ requires com.google.common;
+
+ exports org.restlet.ext.grpc;
+}
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
new file mode 100644
index 0000000000..5a1c208aaf
--- /dev/null
+++ b/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/GrpcClientAdapter.java
@@ -0,0 +1,290 @@
+/**
+ * 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 io.grpc.CallOptions;
+import io.grpc.Channel;
+import io.grpc.ClientCall;
+import io.grpc.ClientInterceptor;
+import io.grpc.ManagedChannel;
+import io.grpc.MethodDescriptor;
+import io.grpc.internal.ManagedChannelImplBuilder;
+import java.time.Duration;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.Executor;
+import java.util.concurrent.TimeUnit;
+import org.restlet.Client;
+import org.restlet.Context;
+import org.restlet.ext.grpc.internal.RestletTransportFactory;
+
+/**
+ * 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.
+ */
+public final class GrpcClientAdapter implements AutoCloseable {
+ private final Context context;
+ private final List interceptors = new CopyOnWriteArrayList<>();
+ private Client client;
+ private String target;
+ private boolean plaintext;
+ private int maxMessageBytes = 1024 * 1024;
+ private int maxConcurrentCalls = 128;
+ private Duration maxCallDuration = Duration.ofSeconds(30);
+ private Duration shutdownTimeout = Duration.ofSeconds(5);
+ private Executor executor;
+ private boolean configured;
+ private ManagedChannel channel;
+ private RestletTransportFactory transportFactory;
+
+ public GrpcClientAdapter() {
+ this(new Context());
+ }
+
+ public GrpcClientAdapter(Context context) {
+ this.context = Objects.requireNonNull(context, "context");
+ }
+
+ public Context getContext() {
+ return context;
+ }
+
+ public synchronized void setClient(Client value) {
+ requireConfigurable();
+ client = Objects.requireNonNull(value, "client");
+ }
+
+ public synchronized Client getClient() {
+ return client;
+ }
+
+ /** Sets a native DNS target, for example {@code dns:///localhost:8080}. */
+ public synchronized void setTarget(String value) {
+ requireConfigurable();
+ target = Objects.requireNonNull(value, "target");
+ }
+
+ public synchronized String getTarget() {
+ return target;
+ }
+
+ public synchronized void setPlaintext(boolean value) {
+ requireConfigurable();
+ plaintext = value;
+ }
+
+ public synchronized boolean isPlaintext() {
+ return plaintext;
+ }
+
+ public synchronized void setMaxMessageBytes(int value) {
+ requireConfigurable();
+ if (value <= 0 || value > Integer.MAX_VALUE - 6) {
+ throw new IllegalArgumentException("A positive finite message limit is required");
+ }
+ maxMessageBytes = value;
+ }
+
+ public synchronized int getMaxMessageBytes() {
+ return maxMessageBytes;
+ }
+
+ public synchronized void setMaxConcurrentCalls(int value) {
+ requireConfigurable();
+ if (value <= 0) {
+ throw new IllegalArgumentException("A positive call limit is required");
+ }
+ maxConcurrentCalls = value;
+ }
+
+ public synchronized int getMaxConcurrentCalls() {
+ return maxConcurrentCalls;
+ }
+
+ public synchronized void setMaxCallDuration(Duration value) {
+ requireConfigurable();
+ maxCallDuration = positive(value);
+ }
+
+ public synchronized Duration getMaxCallDuration() {
+ return maxCallDuration;
+ }
+
+ public synchronized void setShutdownTimeout(Duration value) {
+ requireConfigurable();
+ shutdownTimeout = positive(value);
+ }
+
+ public synchronized Duration getShutdownTimeout() {
+ return shutdownTimeout;
+ }
+
+ /** Sets a borrowed SDK callback executor. The adapter never shuts it down. */
+ public synchronized void setExecutor(Executor value) {
+ requireConfigurable();
+ executor = Objects.requireNonNull(value, "executor");
+ }
+
+ public List getInterceptors() {
+ return interceptors;
+ }
+
+ private static Duration positive(Duration value) {
+ Objects.requireNonNull(value, "duration");
+ if (value.isNegative() || value.isZero()) {
+ throw new IllegalArgumentException("A positive finite duration is required");
+ }
+ return Duration.ofNanos(value.toNanos());
+ }
+
+ private void requireConfigurable() {
+ if (configured) {
+ throw new IllegalStateException("gRPC adapter is single-use; configuration is fixed");
+ }
+ }
+
+ public synchronized void start() {
+ if (isStarted()) {
+ return;
+ }
+ requireConfigurable();
+ if (!plaintext) {
+ throw new UnsupportedOperationException(
+ "R3 supports only explicitly enabled plaintext; TLS is not implemented");
+ }
+ Objects.requireNonNull(target, "target");
+ if (client == null
+ || !client.isStarted()
+ || client.getContext() == null
+ || !"HTTP2"
+ .equals(
+ client.getContext()
+ .getParameters()
+ .getFirstValue("httpClientTransportMode"))
+ || !"false"
+ .equalsIgnoreCase(
+ client.getContext()
+ .getParameters()
+ .getFirstValue("followRedirects", "true"))) {
+ throw new IllegalStateException(
+ "Borrow a started HTTP2 client with followRedirects=false");
+ }
+ if (!target.matches("(dns:///)?[^/?#]+:\\d+")) {
+ throw new IllegalArgumentException("Expected host:port or dns:///host:port target");
+ }
+ configured = true;
+ var factory =
+ new RestletTransportFactory(
+ client, maxMessageBytes, maxConcurrentCalls, shutdownTimeout);
+ try {
+ var builder =
+ new ManagedChannelImplBuilder(target, () -> factory, () -> 80)
+ .disableRetry()
+ .disableServiceConfigLookUp()
+ .proxyDetector(address -> null)
+ .maxInboundMessageSize(maxMessageBytes)
+ .intercept(List.copyOf(interceptors))
+ .intercept(
+ new ClientInterceptor() {
+ @Override
+ public ClientCall interceptCall(
+ MethodDescriptor method,
+ CallOptions options,
+ Channel next) {
+ var cap =
+ io.grpc.Deadline.after(
+ maxCallDuration.toNanos(),
+ TimeUnit.NANOSECONDS);
+ var deadline = options.getDeadline();
+ return next.newCall(
+ method,
+ options.withDeadline(
+ deadline == null
+ ? cap
+ : deadline.minimum(cap)));
+ }
+ });
+ if (executor != null) {
+ builder.executor(executor);
+ }
+ channel = builder.build();
+ transportFactory = factory;
+ } catch (RuntimeException e) {
+ try {
+ factory.close();
+ factory.awaitTermination();
+ } catch (RuntimeException cleanup) {
+ e.addSuppressed(cleanup);
+ }
+ throw e;
+ }
+ }
+
+ public synchronized boolean isStarted() {
+ return channel != null && !channel.isShutdown();
+ }
+
+ public synchronized ManagedChannel getChannel() {
+ if (!isStarted()) {
+ throw new IllegalStateException("gRPC adapter is not started");
+ }
+ return channel;
+ }
+
+ public void stop() {
+ ManagedChannel current;
+ Duration timeout;
+ synchronized (this) {
+ current = channel;
+ timeout = shutdownTimeout;
+ }
+ if (current == null) {
+ return;
+ }
+ current.shutdown();
+ RuntimeException failure = null;
+ try {
+ if (!current.awaitTermination(timeout.toNanos(), TimeUnit.NANOSECONDS)) {
+ current.shutdownNow();
+ if (!current.awaitTermination(timeout.toNanos(), TimeUnit.NANOSECONDS)) {
+ throw new IllegalStateException("gRPC channel shutdown timed out");
+ }
+ }
+ } catch (InterruptedException e) {
+ current.shutdownNow();
+ Thread.currentThread().interrupt();
+ failure = new IllegalStateException("Interrupted awaiting gRPC channel shutdown", e);
+ } catch (RuntimeException e) {
+ failure = e;
+ }
+ try {
+ transportFactory.close();
+ transportFactory.awaitTermination();
+ } catch (RuntimeException e) {
+ if (failure == null) {
+ failure = e;
+ } else {
+ failure.addSuppressed(e);
+ }
+ }
+ if (failure != null) {
+ throw failure;
+ }
+ }
+
+ @Override
+ public void close() {
+ stop();
+ }
+}
diff --git a/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/GuardedClientStream.java b/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/GuardedClientStream.java
new file mode 100644
index 0000000000..e640b9e924
--- /dev/null
+++ b/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/GuardedClientStream.java
@@ -0,0 +1,128 @@
+/**
+ * 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.internal;
+
+import io.grpc.Attributes;
+import io.grpc.Compressor;
+import io.grpc.Deadline;
+import io.grpc.DecompressorRegistry;
+import io.grpc.Status;
+import io.grpc.internal.ClientStream;
+import io.grpc.internal.ClientStreamListener;
+import io.grpc.internal.InsightBuilder;
+import java.io.InputStream;
+
+/** Delayed SDK streams must turn synchronous framing failures into terminal call status. */
+final class GuardedClientStream implements ClientStream {
+ private final ClientStream delegate;
+ private final int maxBytes;
+ private int maxOutboundBytes;
+
+ GuardedClientStream(ClientStream delegate, int maxBytes) {
+ this.delegate = delegate;
+ this.maxBytes = maxBytes;
+ maxOutboundBytes = maxBytes;
+ }
+
+ @Override
+ public void writeMessage(InputStream message) {
+ try {
+ delegate.writeMessage(message);
+ } catch (RuntimeException e) {
+ delegate.cancel(Status.fromThrowable(e));
+ }
+ }
+
+ @Override
+ public void cancel(Status reason) {
+ delegate.cancel(reason);
+ }
+
+ @Override
+ public void halfClose() {
+ delegate.halfClose();
+ }
+
+ @Override
+ public void setAuthority(String authority) {
+ delegate.setAuthority(authority);
+ }
+
+ @Override
+ public void setFullStreamDecompression(boolean enabled) {
+ delegate.setFullStreamDecompression(enabled);
+ }
+
+ @Override
+ public void setDecompressorRegistry(DecompressorRegistry registry) {
+ delegate.setDecompressorRegistry(registry);
+ }
+
+ @Override
+ public void start(ClientStreamListener listener) {
+ // The SDK framer accepts this limit only once, after per-call options are applied.
+ delegate.setMaxOutboundMessageSize(maxOutboundBytes);
+ delegate.start(listener);
+ }
+
+ @Override
+ public void setMaxInboundMessageSize(int size) {
+ delegate.setMaxInboundMessageSize(Math.min(size, maxBytes));
+ }
+
+ @Override
+ public void setMaxOutboundMessageSize(int size) {
+ maxOutboundBytes = Math.min(size, maxBytes);
+ }
+
+ @Override
+ public void setDeadline(Deadline deadline) {
+ delegate.setDeadline(deadline);
+ }
+
+ @Override
+ public Attributes getAttributes() {
+ return delegate.getAttributes();
+ }
+
+ @Override
+ public void appendTimeoutInsight(InsightBuilder insight) {
+ delegate.appendTimeoutInsight(insight);
+ }
+
+ @Override
+ public void request(int count) {
+ delegate.request(count);
+ }
+
+ @Override
+ public void flush() {
+ delegate.flush();
+ }
+
+ @Override
+ public boolean isReady() {
+ return delegate.isReady();
+ }
+
+ @Override
+ public void optimizeForDirectExecutor() {
+ delegate.optimizeForDirectExecutor();
+ }
+
+ @Override
+ public void setCompressor(Compressor compressor) {
+ delegate.setCompressor(compressor);
+ }
+
+ @Override
+ public void setMessageCompression(boolean enabled) {
+ delegate.setMessageCompression(enabled);
+ }
+}
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
new file mode 100644
index 0000000000..67edd46a96
--- /dev/null
+++ b/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletClientStream.java
@@ -0,0 +1,427 @@
+/**
+ * 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.internal;
+
+import io.grpc.Attributes;
+import io.grpc.CallOptions;
+import io.grpc.ClientStreamTracer;
+import io.grpc.InternalMetadata;
+import io.grpc.Metadata;
+import io.grpc.MethodDescriptor;
+import io.grpc.Status;
+import io.grpc.internal.AbstractClientStream;
+import io.grpc.internal.Http2ClientStreamTransportState;
+import io.grpc.internal.ReadableBuffers;
+import io.grpc.internal.SerializingExecutor;
+import io.grpc.internal.StatsTraceContext;
+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.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.function.Consumer;
+import org.restlet.Client;
+import org.restlet.Request;
+import org.restlet.Response;
+import org.restlet.data.Header;
+import org.restlet.data.MediaType;
+import org.restlet.data.Method;
+import org.restlet.data.Reference;
+import org.restlet.representation.ByteArrayRepresentation;
+import org.restlet.util.Series;
+
+/** SDK-framed, bounded unary exchange. No gRPC frame/status codec is implemented here. */
+final class RestletClientStream extends AbstractClientStream {
+ 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 ExecutorService io;
+ private final Consumer finished;
+ private final ByteArrayOutputStream outbound = new ByteArrayOutputStream();
+ private final StatsTraceContext stats;
+ private Metadata headers;
+ private volatile Request request;
+ private volatile Thread worker;
+ private volatile boolean done;
+ private boolean sent;
+
+ RestletClientStream(
+ Client client,
+ InetSocketAddress address,
+ String authority,
+ MethodDescriptor method,
+ Metadata headers,
+ CallOptions options,
+ ClientStreamTracer[] tracers,
+ int limit,
+ ExecutorService io,
+ Consumer finished) {
+ this(
+ client,
+ address,
+ authority,
+ method,
+ headers,
+ options,
+ limit,
+ io,
+ finished,
+ StatsTraceContext.newClientContext(tracers, Attributes.EMPTY, headers),
+ TransportTracer.getDefaultFactory().create());
+ }
+
+ private RestletClientStream(
+ Client client,
+ InetSocketAddress address,
+ String authority,
+ MethodDescriptor method,
+ Metadata headers,
+ CallOptions options,
+ int limit,
+ ExecutorService io,
+ Consumer finished,
+ StatsTraceContext stats,
+ TransportTracer tracer) {
+ super(
+ hint -> new Buffer(Math.max(1, Math.min(hint, 16384))),
+ stats,
+ tracer,
+ headers,
+ options,
+ false);
+ this.client = client;
+ this.authority = authority;
+ String ip = address.getAddress().getHostAddress();
+ if (ip.contains(":")) {
+ ip = "[" + ip + "]";
+ }
+ target = "http://" + ip + ":" + address.getPort() + "/" + method.getFullMethodName();
+ wireLimit = limit + 5;
+ this.io = io;
+ this.finished = finished;
+ this.stats = stats;
+ state = new State(limit, stats, tracer, options);
+ }
+
+ @Override
+ protected State transportState() {
+ return state;
+ }
+
+ @Override
+ public Attributes getAttributes() {
+ return Attributes.EMPTY;
+ }
+
+ @Override
+ public void setAuthority(String value) {
+ if (!authority.equals(value)) {
+ throw new UnsupportedOperationException("Per-call authority override is not supported");
+ }
+ }
+
+ @Override
+ protected Sink abstractClientStreamSink() {
+ return new Sink() {
+ @Override
+ public void writeHeaders(Metadata metadata, byte[] payload) {
+ headers = metadata;
+ stats.clientOutboundHeaders();
+ serial.execute(state::allocated);
+ }
+
+ @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);
+ }
+ }
+
+ @Override
+ public void cancel(Status status) {
+ fail(status);
+ }
+ };
+ }
+
+ private void fail(Status status) {
+ serial.execute(
+ () -> {
+ if (!done) {
+ state.fail(status);
+ }
+ });
+ Request active = request;
+ if (active != null) {
+ active.abort();
+ }
+ Thread activeWorker = worker;
+ if (activeWorker != null && activeWorker != Thread.currentThread()) {
+ activeWorker.interrupt();
+ }
+ }
+
+ private void exchange() {
+ worker = Thread.currentThread();
+ Response response = null;
+ try {
+ 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;
+ if (done || Thread.currentThread().isInterrupted()) {
+ call.abort();
+ return;
+ }
+ call.getDisconnectSignal()
+ .thenAccept(
+ cause ->
+ fail(
+ Status.UNAVAILABLE
+ .withDescription("gRPC peer disconnected")
+ .withCause(cause)));
+ 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));
+ }
+ }
+ }
+ 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 {
+ if (response != null && response.getEntity() != null) {
+ response.getEntity().release();
+ }
+ worker = null;
+ }
+ }
+
+ private static Metadata metadata(Series headers) {
+ var values = new ArrayList();
+ for (Header header : headers) {
+ values.add(
+ header.getName()
+ .toLowerCase(java.util.Locale.ROOT)
+ .getBytes(StandardCharsets.US_ASCII));
+ values.add(header.getValue().getBytes(StandardCharsets.US_ASCII));
+ }
+ return InternalMetadata.newMetadata(
+ TransportFrameUtil.toRawSerializedHeaders(values.toArray(byte[][]::new)));
+ }
+
+ private final class State extends Http2ClientStreamTransportState {
+ private boolean started;
+ private Status pendingFailure;
+
+ private State(
+ int limit, StatsTraceContext stats, TransportTracer tracer, CallOptions options) {
+ super(limit, stats, tracer, options);
+ }
+
+ private void allocated() {
+ started = true;
+ if (!done) {
+ if (pendingFailure != null) {
+ transportReportStatus(pendingFailure, true, new Metadata());
+ } else {
+ onStreamAllocated();
+ }
+ }
+ }
+
+ private void fail(Status status) {
+ if (started) {
+ transportReportStatus(status, true, new Metadata());
+ } else if (pendingFailure == null) {
+ // Shutdown may arrive after stream creation but before SDK listener installation.
+ pendingFailure = status;
+ }
+ }
+
+ private void headers(Metadata metadata) {
+ if (!done) {
+ transportHeadersReceived(metadata);
+ }
+ }
+
+ private void data(byte[] bytes) {
+ if (!done) {
+ transportDataReceived(ReadableBuffers.wrap(bytes), false);
+ }
+ }
+
+ private void trailers(Metadata metadata) {
+ if (!done) {
+ transportTrailersReceived(metadata);
+ }
+ }
+
+ @Override
+ public void runOnTransportThread(Runnable runnable) {
+ serial.execute(runnable);
+ }
+
+ @Override
+ public void bytesRead(int count) {
+ // Unary wire input is bounded and materialized before SDK deframing.
+ }
+
+ @Override
+ public void deframeFailed(Throwable cause) {
+ fail(Status.fromThrowable(cause));
+ }
+
+ @Override
+ protected void http2ProcessingFailed(
+ Status status, boolean stopDelivery, Metadata trailers) {
+ transportReportStatus(status, stopDelivery, trailers);
+ }
+
+ @Override
+ public void deframerClosed(boolean partial) {
+ try {
+ super.deframerClosed(partial);
+ } finally {
+ if (!done) {
+ done = true;
+ finished.accept(RestletClientStream.this);
+ }
+ }
+ }
+ }
+
+ private static final class Buffer implements WritableBuffer {
+ private final byte[] data;
+ private int size;
+
+ private Buffer(int capacity) {
+ data = new byte[capacity];
+ }
+
+ @Override
+ public void write(byte[] source, int offset, int length) {
+ System.arraycopy(source, offset, data, size, length);
+ size += length;
+ }
+
+ @Override
+ public void write(byte value) {
+ data[size++] = value;
+ }
+
+ @Override
+ public int writableBytes() {
+ return data.length - size;
+ }
+
+ @Override
+ public int readableBytes() {
+ return size;
+ }
+
+ @Override
+ public void release() {
+ Arrays.fill(data, (byte) 0);
+ }
+ }
+}
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
new file mode 100644
index 0000000000..19578d1d2e
--- /dev/null
+++ b/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletTransportFactory.java
@@ -0,0 +1,262 @@
+/**
+ * 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.internal;
+
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import io.grpc.Attributes;
+import io.grpc.CallOptions;
+import io.grpc.ChannelCredentials;
+import io.grpc.ChannelLogger;
+import io.grpc.ClientStreamTracer;
+import io.grpc.InternalChannelz;
+import io.grpc.InternalLogId;
+import io.grpc.Metadata;
+import io.grpc.MethodDescriptor;
+import io.grpc.Status;
+import io.grpc.internal.ClientStream;
+import io.grpc.internal.ClientTransportFactory;
+import io.grpc.internal.ConnectionClientTransport;
+import io.grpc.internal.FailingClientStream;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.time.Duration;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import org.restlet.Client;
+
+/** Pinned grpc-java internal SPI. A logical transport is not a physical Netty connection. */
+public final class RestletTransportFactory implements ClientTransportFactory {
+ private final Client client;
+ private final int maxBytes;
+ private final Duration shutdownTimeout;
+ private final Semaphore admission;
+ private final Set transports = new HashSet<>();
+ private final ExecutorService io = Executors.newVirtualThreadPerTaskExecutor();
+ private final ScheduledExecutorService timer =
+ Executors.newSingleThreadScheduledExecutor(
+ Thread.ofPlatform().daemon().name("restlet-grpc-timer").factory());
+ private boolean closed;
+
+ public RestletTransportFactory(Client client, int maxBytes, int maxCalls, Duration timeout) {
+ this.client = client;
+ this.maxBytes = maxBytes;
+ admission = new Semaphore(maxCalls);
+ shutdownTimeout = timeout;
+ }
+
+ @Override
+ public synchronized ConnectionClientTransport newClientTransport(
+ SocketAddress address, ClientTransportOptions options, ChannelLogger logger) {
+ if (closed) {
+ throw new IllegalStateException("gRPC transport factory is closed");
+ }
+ if (!(address instanceof InetSocketAddress inet)
+ || inet.isUnresolved()
+ || options.getHttpConnectProxiedSocketAddress() != null) {
+ throw new IllegalArgumentException(
+ "Only resolved Internet addresses without proxy CONNECT are supported");
+ }
+ Transport transport = new Transport(inet, options.getAuthority());
+ transports.add(transport);
+ return transport;
+ }
+
+ @Override
+ public ScheduledExecutorService getScheduledExecutorService() {
+ return timer;
+ }
+
+ @Override
+ public SwapChannelCredentialsResult swapChannelCredentials(ChannelCredentials credentials) {
+ throw new UnsupportedOperationException("Channel credentials/TLS are not supported in R3");
+ }
+
+ @Override
+ public Collection> getSupportedSocketAddressTypes() {
+ return Set.of(InetSocketAddress.class);
+ }
+
+ @Override
+ public void close() {
+ Set snapshot;
+ synchronized (this) {
+ if (closed) {
+ return;
+ }
+ closed = true;
+ snapshot = Set.copyOf(transports);
+ }
+ snapshot.forEach(t -> t.shutdownNow(Status.UNAVAILABLE.withDescription("Adapter closed")));
+ timer.shutdownNow();
+ io.shutdownNow();
+ }
+
+ /** Called by the facade, never from the SDK transport callback that may close the factory. */
+ public void awaitTermination() {
+ try {
+ if (!io.awaitTermination(shutdownTimeout.toNanos(), TimeUnit.NANOSECONDS)
+ || !timer.awaitTermination(shutdownTimeout.toNanos(), TimeUnit.NANOSECONDS)) {
+ throw new IllegalStateException("gRPC transport worker shutdown timed out");
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException("Interrupted closing gRPC transport workers", e);
+ }
+ }
+
+ private final class Transport implements ConnectionClientTransport {
+ private final InetSocketAddress address;
+ private final String authority;
+ private final Set streams = new HashSet<>();
+ private final InternalLogId id = InternalLogId.allocate(Transport.class, "Restlet HTTP/2");
+ private Listener listener;
+ private Status shutdown;
+ private boolean terminated;
+
+ private Transport(InetSocketAddress address, String authority) {
+ this.address = address;
+ this.authority = authority;
+ }
+
+ @Override
+ public synchronized Runnable start(Listener value) {
+ listener = value;
+ return () -> {
+ synchronized (Transport.this) {
+ if (shutdown == null) {
+ listener.transportReady();
+ }
+ }
+ };
+ }
+
+ @Override
+ public synchronized ClientStream newStream(
+ MethodDescriptor method,
+ Metadata headers,
+ CallOptions options,
+ ClientStreamTracer[] tracers) {
+ Status rejected = shutdown;
+ if (rejected == null
+ && options.getAuthority() != null
+ && !authority.equals(options.getAuthority())) {
+ rejected =
+ 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
+ && options.getCompressor() != null
+ && !"identity".equals(options.getCompressor())) {
+ rejected =
+ Status.UNIMPLEMENTED.withDescription(
+ "Request compression is not supported");
+ }
+ if (rejected != null) {
+ return new FailingClientStream(rejected, tracers);
+ }
+ if (!admission.tryAcquire()) {
+ return new FailingClientStream(
+ Status.RESOURCE_EXHAUSTED.withDescription("gRPC call limit exceeded"),
+ tracers);
+ }
+ RestletClientStream stream =
+ new RestletClientStream(
+ client,
+ address,
+ authority,
+ method,
+ headers,
+ options,
+ tracers,
+ maxBytes,
+ io,
+ this::finished);
+ streams.add(stream);
+ if (streams.size() == 1) {
+ listener.transportInUse(true);
+ }
+ return new GuardedClientStream(stream, maxBytes);
+ }
+
+ private synchronized void finished(RestletClientStream stream) {
+ if (streams.remove(stream)) {
+ admission.release();
+ if (streams.isEmpty()) {
+ listener.transportInUse(false);
+ terminateIfIdle();
+ }
+ }
+ }
+
+ @Override
+ public synchronized void shutdown(Status status) {
+ if (shutdown == null) {
+ shutdown = status;
+ listener.transportShutdown(status, null);
+ terminateIfIdle();
+ }
+ }
+
+ private void terminateIfIdle() {
+ if (shutdown != null && streams.isEmpty() && !terminated) {
+ terminated = true;
+ listener.transportTerminated();
+ synchronized (RestletTransportFactory.this) {
+ transports.remove(this);
+ }
+ }
+ }
+
+ @Override
+ public void shutdownNow(Status status) {
+ Set active;
+ synchronized (this) {
+ shutdown(status);
+ active = Set.copyOf(streams);
+ }
+ active.forEach(s -> s.cancel(status));
+ }
+
+ @Override
+ public void ping(PingCallback callback, Executor executor) {
+ executor.execute(
+ () ->
+ callback.onFailure(
+ Status.UNIMPLEMENTED.withDescription(
+ "Physical connection PING is not supported")));
+ }
+
+ @Override
+ public ListenableFuture getStats() {
+ // Restlet owns the pool; exposing fabricated per-socket counters would be misleading.
+ return Futures.immediateFuture(null);
+ }
+
+ @Override
+ public InternalLogId getLogId() {
+ return id;
+ }
+
+ @Override
+ public Attributes getAttributes() {
+ return Attributes.EMPTY;
+ }
+ }
+}
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
new file mode 100644
index 0000000000..38f53b321e
--- /dev/null
+++ b/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcClientAdapterTestCase.java
@@ -0,0 +1,759 @@
+/**
+ * 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.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import com.google.protobuf.DynamicMessage;
+import com.google.protobuf.StringValue;
+import io.grpc.CallOptions;
+import io.grpc.ClientCall;
+import io.grpc.Metadata;
+import io.grpc.MethodDescriptor;
+import io.grpc.ServerCall;
+import io.grpc.ServerCallHandler;
+import io.grpc.ServerInterceptor;
+import io.grpc.ServerServiceDefinition;
+import io.grpc.Status;
+import io.grpc.StatusRuntimeException;
+import io.grpc.health.v1.HealthCheckRequest;
+import io.grpc.health.v1.HealthCheckResponse;
+import io.grpc.health.v1.HealthGrpc;
+import io.grpc.netty.shaded.io.grpc.netty.NettyServerBuilder;
+import io.grpc.protobuf.ProtoUtils;
+import io.grpc.protobuf.services.HealthStatusManager;
+import io.grpc.stub.ClientCalls;
+import io.grpc.stub.ServerCalls;
+import java.time.Duration;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import org.restlet.Client;
+import org.restlet.Context;
+import org.restlet.data.Protocol;
+
+class GrpcClientAdapterTestCase {
+ private static final MethodDescriptor ECHO =
+ MethodDescriptor.newBuilder()
+ .setType(MethodDescriptor.MethodType.UNARY)
+ .setFullMethodName("review.Unary/Echo")
+ .setRequestMarshaller(ProtoUtils.marshaller(StringValue.getDefaultInstance()))
+ .setResponseMarshaller(ProtoUtils.marshaller(StringValue.getDefaultInstance()))
+ .build();
+ private static final Metadata.Key TEXT =
+ Metadata.Key.of("x-review", Metadata.ASCII_STRING_MARSHALLER);
+ private static final Metadata.Key BINARY =
+ Metadata.Key.of("grpc-status-details-bin", Metadata.BINARY_BYTE_MARSHALLER);
+ private final GrpcClientAdapter adapter = new GrpcClientAdapter();
+ private Client client;
+ private io.grpc.Server peer;
+
+ private void start(ServerServiceDefinition service, ServerInterceptor... interceptors)
+ throws Exception {
+ start(service, "dns:///localhost", interceptors);
+ }
+
+ private void start(
+ ServerServiceDefinition service, String targetHost, ServerInterceptor... interceptors)
+ throws Exception {
+ var builder =
+ NettyServerBuilder.forPort(0)
+ .addService(service)
+ .addService(new HealthStatusManager().getHealthService());
+ for (var interceptor : interceptors) {
+ builder.intercept(interceptor);
+ }
+ peer = builder.build().start();
+ 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(targetHost + ":" + peer.getPort());
+ adapter.setPlaintext(true);
+ adapter.start();
+ }
+
+ private static ServerServiceDefinition echo() {
+ return ServerServiceDefinition.builder("review.Unary")
+ .addMethod(
+ ECHO,
+ ServerCalls.asyncUnaryCall(
+ (StringValue request,
+ io.grpc.stub.StreamObserver out) -> {
+ if ("bad".equals(request.getValue())) {
+ Metadata trailers = new Metadata();
+ trailers.put(TEXT, "one");
+ trailers.put(TEXT, "two");
+ trailers.put(BINARY, new byte[] {0, (byte) 255, 7});
+ out.onError(
+ Status.INVALID_ARGUMENT
+ .withDescription(
+ "native % error: caf\u00e9")
+ .asRuntimeException(trailers));
+ } else {
+ out.onNext(request);
+ out.onCompleted();
+ }
+ }))
+ .build();
+ }
+
+ @AfterEach
+ void cleanup() throws Exception {
+ try {
+ adapter.stop();
+ } finally {
+ try {
+ if (client != null) {
+ client.stop();
+ }
+ } finally {
+ if (peer != null) {
+ peer.shutdownNow();
+ assertTrue(peer.awaitTermination(30, TimeUnit.SECONDS));
+ }
+ }
+ }
+ }
+
+ private StringValue call(String value) {
+ return ClientCalls.blockingUnaryCall(
+ adapter.getChannel(), ECHO, CallOptions.DEFAULT, StringValue.of(value));
+ }
+
+ @Test
+ void generatedAndDynamicUnaryCallsUseOfficialSdk() throws Exception {
+ start(echo());
+ assertEquals("hello", call("hello").getValue());
+ var health =
+ HealthGrpc.newBlockingStub(adapter.getChannel())
+ .check(HealthCheckRequest.getDefaultInstance());
+ assertEquals(HealthCheckResponse.ServingStatus.SERVING, health.getStatus());
+ var dynamic =
+ MethodDescriptor.newBuilder()
+ .setType(MethodDescriptor.MethodType.UNARY)
+ .setFullMethodName(HealthGrpc.getCheckMethod().getFullMethodName())
+ .setRequestMarshaller(
+ ProtoUtils.marshaller(
+ DynamicMessage.getDefaultInstance(
+ HealthCheckRequest.getDescriptor())))
+ .setResponseMarshaller(
+ ProtoUtils.marshaller(
+ DynamicMessage.getDefaultInstance(
+ HealthCheckResponse.getDescriptor())))
+ .build();
+ DynamicMessage result =
+ ClientCalls.blockingUnaryCall(
+ adapter.getChannel(),
+ dynamic,
+ CallOptions.DEFAULT,
+ DynamicMessage.getDefaultInstance(HealthCheckRequest.getDescriptor()));
+ assertEquals(
+ health.getStatus().getNumber(),
+ ((com.google.protobuf.Descriptors.EnumValueDescriptor)
+ result.getField(
+ HealthCheckResponse.getDescriptor()
+ .findFieldByName("status")))
+ .getNumber());
+ assertTrue(adapter.getChannel().authority().startsWith("localhost:"));
+ adapter.stop();
+ assertTrue(client.isStarted());
+ assertFalse(adapter.isStarted());
+ assertThrows(IllegalStateException.class, adapter::getChannel);
+ assertThrows(IllegalStateException.class, adapter::start);
+ }
+
+ @Test
+ void preservesTrailersOnlyErrorAndRecovers() throws Exception {
+ start(echo());
+ StatusRuntimeException error =
+ assertThrows(StatusRuntimeException.class, () -> call("bad"));
+ assertEquals(Status.Code.INVALID_ARGUMENT, error.getStatus().getCode());
+ assertEquals("native % error: caf\u00e9", error.getStatus().getDescription());
+ assertEquals(
+ List.of("one", "two"),
+ com.google.common.collect.ImmutableList.copyOf(error.getTrailers().getAll(TEXT)));
+ assertArrayEquals(new byte[] {0, (byte) 255, 7}, error.getTrailers().get(BINARY));
+ assertEquals("recovered", call("recovered").getValue());
+ }
+
+ @Test
+ void passesMetadataAndAuthorityWithoutLosingResolvedPeer() throws Exception {
+ AtomicReference authority = new AtomicReference<>();
+ AtomicReference observed = new AtomicReference<>();
+ start(
+ echo(),
+ new ServerInterceptor() {
+ @Override
+ public ServerCall.Listener interceptCall(
+ ServerCall call, Metadata headers, ServerCallHandler next) {
+ authority.set(call.getAuthority());
+ observed.set(headers);
+ return next.startCall(call, headers);
+ }
+ });
+ Metadata metadata = new Metadata();
+ metadata.put(TEXT, "first");
+ metadata.put(TEXT, "second");
+ metadata.put(BINARY, new byte[] {1, 2, (byte) 255});
+ metadata.put(BINARY, new byte[] {(byte) 254, 0});
+ var channel =
+ io.grpc.ClientInterceptors.intercept(
+ adapter.getChannel(),
+ io.grpc.stub.MetadataUtils.newAttachHeadersInterceptor(metadata));
+ assertEquals(
+ "meta",
+ ClientCalls.blockingUnaryCall(
+ channel, ECHO, CallOptions.DEFAULT, StringValue.of("meta"))
+ .getValue());
+ assertEquals("localhost:" + peer.getPort(), authority.get());
+ assertEquals(
+ List.of("first", "second"),
+ com.google.common.collect.ImmutableList.copyOf(observed.get().getAll(TEXT)));
+ var binary = com.google.common.collect.ImmutableList.copyOf(observed.get().getAll(BINARY));
+ assertEquals(2, binary.size());
+ assertArrayEquals(new byte[] {1, 2, (byte) 255}, binary.get(0));
+ assertArrayEquals(new byte[] {(byte) 254, 0}, binary.get(1));
+ }
+
+ @Test
+ void bareIpTargetPreservesAuthorityAndAcceptsMatchingCallAuthority() throws Exception {
+ AtomicReference authority = new AtomicReference<>();
+ start(
+ echo(),
+ "127.0.0.1",
+ new ServerInterceptor() {
+ @Override
+ public ServerCall.Listener interceptCall(
+ ServerCall call, Metadata headers, ServerCallHandler next) {
+ authority.set(call.getAuthority());
+ return next.startCall(call, headers);
+ }
+ });
+ String expected = "127.0.0.1:" + peer.getPort();
+ assertEquals(expected, adapter.getChannel().authority());
+ assertEquals(
+ "ip",
+ ClientCalls.blockingUnaryCall(
+ adapter.getChannel(),
+ ECHO,
+ CallOptions.DEFAULT.withAuthority(expected),
+ StringValue.of("ip"))
+ .getValue());
+ assertEquals(expected, authority.get());
+ }
+
+ @Test
+ void cancelledBeforeStartNeverReachesPeerAndHasOneTerminalCallback() throws Exception {
+ AtomicInteger requests = new AtomicInteger();
+ start(
+ echo(),
+ new ServerInterceptor() {
+ @Override
+ public ServerCall.Listener interceptCall(
+ ServerCall call, Metadata headers, ServerCallHandler next) {
+ requests.incrementAndGet();
+ return next.startCall(call, headers);
+ }
+ });
+ ClientCall call =
+ adapter.getChannel().newCall(ECHO, CallOptions.DEFAULT);
+ AtomicInteger terminals = new AtomicInteger();
+ AtomicReference status = new AtomicReference<>();
+ CountDownLatch closed = new CountDownLatch(1);
+ call.cancel("cancelled before start", null);
+ call.start(
+ new ClientCall.Listener<>() {
+ @Override
+ public void onClose(Status value, Metadata trailers) {
+ status.set(value);
+ terminals.incrementAndGet();
+ closed.countDown();
+ }
+ },
+ new Metadata());
+ assertTrue(closed.await(30, TimeUnit.SECONDS));
+ assertEquals(Status.Code.CANCELLED, status.get().getCode());
+ assertEquals(0, requests.get());
+ assertEquals("next", call("next").getValue());
+ assertEquals(1, requests.get());
+ assertEquals(1, terminals.get());
+ }
+
+ @Test
+ void cancellationAfterHeadersReleasesAdmissionAndClosesExactlyOnce() throws Exception {
+ adapter.setMaxConcurrentCalls(1);
+ CountDownLatch cancelled = new CountDownLatch(1);
+ start(
+ ServerServiceDefinition.builder("review.Unary")
+ .addMethod(
+ // Flush one response without completing the client's unary call.
+ ECHO.toBuilder()
+ .setType(MethodDescriptor.MethodType.SERVER_STREAMING)
+ .build(),
+ ServerCalls.asyncServerStreamingCall(
+ (StringValue request,
+ io.grpc.stub.StreamObserver out) -> {
+ if ("block".equals(request.getValue())) {
+ ((io.grpc.stub.ServerCallStreamObserver<
+ StringValue>)
+ out)
+ .setOnCancelHandler(cancelled::countDown);
+ out.onNext(StringValue.of("pending"));
+ } else {
+ out.onNext(request);
+ out.onCompleted();
+ }
+ }))
+ .build());
+ CountDownLatch headers = new CountDownLatch(1);
+ CountDownLatch closed = new CountDownLatch(1);
+ AtomicInteger terminals = new AtomicInteger();
+ AtomicInteger messages = new AtomicInteger();
+ AtomicReference status = new AtomicReference<>();
+ ClientCall call =
+ adapter.getChannel().newCall(ECHO, CallOptions.DEFAULT);
+ call.start(
+ new ClientCall.Listener<>() {
+ @Override
+ public void onHeaders(Metadata value) {
+ headers.countDown();
+ }
+
+ @Override
+ public void onMessage(StringValue value) {
+ messages.incrementAndGet();
+ }
+
+ @Override
+ public void onClose(Status value, Metadata trailers) {
+ status.set(value);
+ terminals.incrementAndGet();
+ closed.countDown();
+ }
+ },
+ new Metadata());
+ call.request(1);
+ call.sendMessage(StringValue.of("block"));
+ call.halfClose();
+ assertTrue(headers.await(30, TimeUnit.SECONDS));
+ call.cancel("cancel after headers", null);
+ call.cancel("duplicate", null);
+ assertTrue(closed.await(30, TimeUnit.SECONDS));
+ assertTrue(cancelled.await(30, TimeUnit.SECONDS));
+ assertEquals(Status.Code.CANCELLED, status.get().getCode());
+ assertEquals("recovered", call("recovered").getValue());
+ assertEquals(1, terminals.get());
+ assertEquals(0, messages.get());
+ }
+
+ private record Pending(CountDownLatch entered, CountDownLatch cancelled) {}
+
+ private Pending startBlocked() throws Exception {
+ CountDownLatch entered = new CountDownLatch(1);
+ CountDownLatch cancelled = new CountDownLatch(1);
+ var service =
+ ServerServiceDefinition.builder("review.Unary")
+ .addMethod(
+ ECHO,
+ ServerCalls.asyncUnaryCall(
+ (StringValue request,
+ io.grpc.stub.StreamObserver out) -> {
+ if ("block".equals(request.getValue())) {
+ ((io.grpc.stub.ServerCallStreamObserver<
+ StringValue>)
+ out)
+ .setOnCancelHandler(cancelled::countDown);
+ entered.countDown();
+ } else {
+ out.onNext(request);
+ out.onCompleted();
+ }
+ }))
+ .build();
+ start(service);
+ return new Pending(entered, cancelled);
+ }
+
+ @Test
+ void cancellationBeforeHeadersHasOneTerminalCallbackAndRecovery() throws Exception {
+ Pending pending = startBlocked();
+ AtomicInteger terminals = new AtomicInteger();
+ CountDownLatch closed = new CountDownLatch(1);
+ AtomicReference status = new AtomicReference<>();
+ ClientCall call =
+ adapter.getChannel().newCall(ECHO, CallOptions.DEFAULT);
+ call.start(
+ new ClientCall.Listener<>() {
+ @Override
+ public void onClose(Status value, Metadata trailers) {
+ status.set(value);
+ terminals.incrementAndGet();
+ closed.countDown();
+ }
+ },
+ new Metadata());
+ call.request(1);
+ call.sendMessage(StringValue.of("block"));
+ call.halfClose();
+ assertTrue(pending.entered().await(30, TimeUnit.SECONDS));
+ call.cancel("review cancellation", null);
+ call.cancel("duplicate cancellation", null);
+ assertTrue(closed.await(30, TimeUnit.SECONDS));
+ assertTrue(pending.cancelled().await(30, TimeUnit.SECONDS));
+ assertEquals(Status.Code.CANCELLED, status.get().getCode());
+ assertEquals("next", call("next").getValue());
+ assertEquals(1, terminals.get());
+ }
+
+ @Test
+ void deadlineAndConcurrentAdmissionAreNativeFailures() throws Exception {
+ adapter.setMaxConcurrentCalls(1);
+ Pending pending = startBlocked();
+ var first =
+ ClientCalls.futureUnaryCall(
+ adapter.getChannel()
+ .newCall(
+ ECHO,
+ CallOptions.DEFAULT.withDeadlineAfter(3, TimeUnit.SECONDS)),
+ StringValue.of("block"));
+ assertTrue(pending.entered().await(30, TimeUnit.SECONDS));
+ assertEquals(
+ Status.Code.RESOURCE_EXHAUSTED,
+ assertThrows(StatusRuntimeException.class, () -> call("second"))
+ .getStatus()
+ .getCode());
+ var error =
+ assertThrows(
+ java.util.concurrent.ExecutionException.class,
+ () -> first.get(30, TimeUnit.SECONDS));
+ assertEquals(Status.Code.DEADLINE_EXCEEDED, Status.fromThrowable(error).getCode());
+ assertTrue(pending.cancelled().await(30, TimeUnit.SECONDS));
+ assertEquals("recovered", call("recovered").getValue());
+ }
+
+ @Test
+ void stopCancelsActiveCallsAndLeavesBorrowedConnectorRunning() throws Exception {
+ adapter.setShutdownTimeout(Duration.ofMillis(100));
+ Pending pending = startBlocked();
+ var first =
+ ClientCalls.futureUnaryCall(
+ adapter.getChannel().newCall(ECHO, CallOptions.DEFAULT),
+ StringValue.of("block"));
+ assertTrue(pending.entered().await(30, TimeUnit.SECONDS));
+ adapter.stop();
+ assertThrows(
+ java.util.concurrent.ExecutionException.class,
+ () -> first.get(30, TimeUnit.SECONDS));
+ assertTrue(pending.cancelled().await(30, TimeUnit.SECONDS));
+ assertTrue(client.isStarted());
+ }
+
+ @Test
+ void interruptedStopStillClosesOwnedChannelAndPreservesBorrowedResources() throws Exception {
+ Pending pending = startBlocked();
+ var channel = adapter.getChannel();
+ var first =
+ ClientCalls.futureUnaryCall(
+ channel.newCall(ECHO, CallOptions.DEFAULT), StringValue.of("block"));
+ assertTrue(pending.entered().await(30, TimeUnit.SECONDS));
+ try {
+ Thread.currentThread().interrupt();
+ assertThrows(IllegalStateException.class, adapter::stop);
+ assertTrue(Thread.currentThread().isInterrupted());
+ } finally {
+ Thread.interrupted();
+ }
+ assertTrue(channel.awaitTermination(30, TimeUnit.SECONDS));
+ assertThrows(
+ java.util.concurrent.ExecutionException.class,
+ () -> first.get(30, TimeUnit.SECONDS));
+ assertTrue(pending.cancelled().await(30, TimeUnit.SECONDS));
+ assertTrue(client.isStarted());
+ adapter.stop();
+ }
+
+ @Test
+ void rejectsStreamingAndOutboundLimitBeforeSending() throws Exception {
+ adapter.setMaxMessageBytes(128);
+ start(echo());
+ assertEquals(
+ Status.Code.RESOURCE_EXHAUSTED,
+ assertThrows(StatusRuntimeException.class, () -> call("x".repeat(129)))
+ .getStatus()
+ .getCode());
+ assertEquals("small", call("small").getValue());
+ var streamMethod =
+ ECHO.toBuilder().setType(MethodDescriptor.MethodType.SERVER_STREAMING).build();
+ assertEquals(
+ Status.Code.UNIMPLEMENTED,
+ assertThrows(
+ StatusRuntimeException.class,
+ () ->
+ ClientCalls.blockingServerStreamingCall(
+ adapter.getChannel(),
+ streamMethod,
+ CallOptions.DEFAULT,
+ StringValue.of("x"))
+ .next())
+ .getStatus()
+ .getCode());
+ }
+
+ @Test
+ void honorsExactMessageLimitAndStricterCallOptions() throws Exception {
+ adapter.setMaxMessageBytes(128);
+ start(echo());
+ StringValue boundary = StringValue.of("x".repeat(126));
+ StringValue oversized = StringValue.of("x".repeat(127));
+ assertEquals(128, boundary.getSerializedSize());
+ assertEquals(129, oversized.getSerializedSize());
+ for (int limit : List.of(128, 4096)) {
+ CallOptions options =
+ CallOptions.DEFAULT
+ .withMaxInboundMessageSize(limit)
+ .withMaxOutboundMessageSize(limit);
+ assertEquals(
+ boundary,
+ ClientCalls.blockingUnaryCall(adapter.getChannel(), ECHO, options, boundary));
+ assertEquals(
+ Status.Code.RESOURCE_EXHAUSTED,
+ assertThrows(
+ StatusRuntimeException.class,
+ () ->
+ ClientCalls.blockingUnaryCall(
+ adapter.getChannel(), ECHO, options, oversized))
+ .getStatus()
+ .getCode());
+ }
+ for (CallOptions options :
+ List.of(
+ CallOptions.DEFAULT.withMaxInboundMessageSize(64),
+ CallOptions.DEFAULT.withMaxOutboundMessageSize(64))) {
+ assertEquals(
+ Status.Code.RESOURCE_EXHAUSTED,
+ assertThrows(
+ StatusRuntimeException.class,
+ () ->
+ ClientCalls.blockingUnaryCall(
+ adapter.getChannel(), ECHO, options, boundary))
+ .getStatus()
+ .getCode());
+ assertEquals("recovered", call("recovered").getValue());
+ }
+ }
+
+ @Test
+ void boundsInboundResponseAndDoesNotAllowCallOptionsToRaiseLimits() throws Exception {
+ adapter.setMaxMessageBytes(128);
+ var service =
+ ServerServiceDefinition.builder("review.Unary")
+ .addMethod(
+ ECHO,
+ ServerCalls.asyncUnaryCall(
+ (StringValue request,
+ io.grpc.stub.StreamObserver out) -> {
+ out.onNext(StringValue.of("x".repeat(256)));
+ out.onCompleted();
+ }))
+ .build();
+ start(service);
+ assertEquals(
+ Status.Code.RESOURCE_EXHAUSTED,
+ assertThrows(
+ StatusRuntimeException.class,
+ () ->
+ ClientCalls.blockingUnaryCall(
+ adapter.getChannel(),
+ ECHO,
+ CallOptions.DEFAULT.withMaxInboundMessageSize(4096),
+ StringValue.of("large")))
+ .getStatus()
+ .getCode());
+ assertEquals(
+ Status.Code.RESOURCE_EXHAUSTED,
+ assertThrows(
+ StatusRuntimeException.class,
+ () ->
+ ClientCalls.blockingUnaryCall(
+ adapter.getChannel(),
+ ECHO,
+ CallOptions.DEFAULT.withMaxOutboundMessageSize(
+ 4096),
+ StringValue.of("x".repeat(256))))
+ .getStatus()
+ .getCode());
+ }
+
+ @Test
+ void enforcesDefaultCallDurationAndCallerMayShortenIt() throws Exception {
+ adapter.setMaxCallDuration(Duration.ofSeconds(3));
+ Pending pending = startBlocked();
+ var failure = assertThrows(StatusRuntimeException.class, () -> call("block"));
+ assertEquals(Status.Code.DEADLINE_EXCEEDED, failure.getStatus().getCode());
+ assertTrue(pending.cancelled().await(30, TimeUnit.SECONDS));
+ assertEquals("next", call("next").getValue());
+ }
+
+ @Test
+ void callerDeadlineCanShortenAdapterDuration() throws Exception {
+ adapter.setMaxCallDuration(Duration.ofSeconds(20));
+ Pending pending = startBlocked();
+ var first =
+ ClientCalls.futureUnaryCall(
+ adapter.getChannel()
+ .newCall(
+ ECHO,
+ CallOptions.DEFAULT.withDeadlineAfter(3, TimeUnit.SECONDS)),
+ StringValue.of("block"));
+ assertTrue(pending.entered().await(30, TimeUnit.SECONDS));
+ var error =
+ assertThrows(
+ java.util.concurrent.ExecutionException.class,
+ () -> first.get(10, TimeUnit.SECONDS));
+ assertEquals(Status.Code.DEADLINE_EXCEEDED, Status.fromThrowable(error).getCode());
+ assertTrue(pending.cancelled().await(30, TimeUnit.SECONDS));
+ assertEquals("next", call("next").getValue());
+ }
+
+ @Test
+ void rejectsCompressionAndAuthorityOverridesExplicitly() throws Exception {
+ start(echo());
+ for (CallOptions options :
+ List.of(
+ CallOptions.DEFAULT.withCompression("gzip"),
+ CallOptions.DEFAULT.withAuthority("different.example"))) {
+ assertEquals(
+ Status.Code.UNIMPLEMENTED,
+ assertThrows(
+ StatusRuntimeException.class,
+ () ->
+ ClientCalls.blockingUnaryCall(
+ adapter.getChannel(),
+ ECHO,
+ options,
+ StringValue.of("x")))
+ .getStatus()
+ .getCode());
+ }
+ assertEquals("next", call("next").getValue());
+ }
+
+ @Test
+ void remoteShutdownTerminatesCallWithoutStoppingBorrowedConnector() throws Exception {
+ adapter.setMaxConcurrentCalls(1);
+ Pending pending = startBlocked();
+ int port = peer.getPort();
+ var call =
+ ClientCalls.futureUnaryCall(
+ adapter.getChannel().newCall(ECHO, CallOptions.DEFAULT),
+ StringValue.of("block"));
+ assertTrue(pending.entered().await(30, TimeUnit.SECONDS));
+ peer.shutdownNow();
+ var error =
+ assertThrows(
+ java.util.concurrent.ExecutionException.class,
+ () -> call.get(30, TimeUnit.SECONDS));
+ assertEquals(Status.Code.UNAVAILABLE, Status.fromThrowable(error).getCode());
+ assertTrue(client.isStarted());
+ assertTrue(peer.awaitTermination(30, TimeUnit.SECONDS));
+ peer = NettyServerBuilder.forPort(port).addService(echo()).build().start();
+ assertEquals("reconnected", call("reconnected").getValue());
+ }
+
+ @Test
+ void stoppingOneAdapterLeavesSharedConnectorAndOtherChannelUsable() throws Exception {
+ start(echo());
+ try (var other = new GrpcClientAdapter()) {
+ other.setClient(client);
+ other.setTarget(adapter.getTarget());
+ other.setPlaintext(true);
+ other.start();
+ assertEquals("first", call("first").getValue());
+ adapter.stop();
+ adapter.stop();
+ assertTrue(client.isStarted());
+ assertEquals(
+ "second",
+ ClientCalls.blockingUnaryCall(
+ other.getChannel(),
+ ECHO,
+ CallOptions.DEFAULT,
+ StringValue.of("second"))
+ .getValue());
+ }
+ assertTrue(client.isStarted());
+ }
+
+ @Test
+ void nativeInterceptorAndBorrowedExecutorArePreserved() throws Exception {
+ try (var callbacks = java.util.concurrent.Executors.newVirtualThreadPerTaskExecutor()) {
+ adapter.setExecutor(callbacks);
+ AtomicInteger intercepted = new AtomicInteger();
+ adapter.getInterceptors()
+ .add(
+ new io.grpc.ClientInterceptor() {
+ @Override
+ public ClientCall interceptCall(
+ MethodDescriptor method,
+ CallOptions options,
+ io.grpc.Channel next) {
+ intercepted.incrementAndGet();
+ return next.newCall(method, options);
+ }
+ });
+ start(echo());
+ adapter.start();
+ assertEquals("one", call("one").getValue());
+ adapter.getInterceptors().clear();
+ assertEquals("two", call("two").getValue());
+ assertEquals(2, intercepted.get());
+ assertThrows(IllegalStateException.class, () -> adapter.setMaxMessageBytes(256));
+ adapter.stop();
+ assertFalse(callbacks.isShutdown());
+ assertSame(client, adapter.getClient());
+ }
+ }
+
+ @Test
+ void validatesConfigurationAndRequiresExplicitPlaintext() {
+ assertFalse(adapter.isStarted());
+ assertThrows(IllegalStateException.class, adapter::getChannel);
+ assertThrows(UnsupportedOperationException.class, adapter::start);
+ assertThrows(IllegalArgumentException.class, () -> adapter.setMaxMessageBytes(0));
+ assertThrows(IllegalArgumentException.class, () -> adapter.setMaxConcurrentCalls(0));
+ assertThrows(
+ IllegalArgumentException.class, () -> adapter.setMaxCallDuration(Duration.ZERO));
+ assertThrows(
+ IllegalArgumentException.class, () -> adapter.setShutdownTimeout(Duration.ZERO));
+ adapter.setPlaintext(true);
+ assertThrows(NullPointerException.class, adapter::start);
+ adapter.setTarget("dns:///localhost:8080");
+ assertThrows(IllegalStateException.class, adapter::start);
+ assertNotNull(adapter.getContext());
+ assertEquals(128, adapter.getMaxConcurrentCalls());
+ assertEquals(1024 * 1024, adapter.getMaxMessageBytes());
+ assertEquals(Duration.ofSeconds(30), adapter.getMaxCallDuration());
+ assertEquals(Duration.ofSeconds(5), adapter.getShutdownTimeout());
+ assertEquals("dns:///localhost:8080", adapter.getTarget());
+ assertTrue(adapter.isPlaintext());
+ }
+}
diff --git a/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcHttpValidationTestCase.java b/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcHttpValidationTestCase.java
new file mode 100644
index 0000000000..c285658103
--- /dev/null
+++ b/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcHttpValidationTestCase.java
@@ -0,0 +1,283 @@
+/**
+ * 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.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import com.google.protobuf.StringValue;
+import io.grpc.CallOptions;
+import io.grpc.MethodDescriptor;
+import io.grpc.Status;
+import io.grpc.StatusRuntimeException;
+import io.grpc.protobuf.ProtoUtils;
+import io.grpc.stub.ClientCalls;
+import java.nio.ByteBuffer;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Consumer;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import org.restlet.Client;
+import org.restlet.Context;
+import org.restlet.Request;
+import org.restlet.Response;
+import org.restlet.Restlet;
+import org.restlet.Server;
+import org.restlet.data.MediaType;
+import org.restlet.data.Protocol;
+import org.restlet.representation.ByteArrayRepresentation;
+
+class GrpcHttpValidationTestCase {
+ private final GrpcClientAdapter adapter = new GrpcClientAdapter();
+ private Server peer;
+ private Client client;
+ private static final MethodDescriptor METHOD =
+ MethodDescriptor.newBuilder()
+ .setType(MethodDescriptor.MethodType.UNARY)
+ .setFullMethodName("review.Invalid/Call")
+ .setRequestMarshaller(ProtoUtils.marshaller(StringValue.getDefaultInstance()))
+ .setResponseMarshaller(ProtoUtils.marshaller(StringValue.getDefaultInstance()))
+ .build();
+
+ private void start(Restlet handler) throws Exception {
+ Context serverContext = new Context();
+ serverContext.getParameters().set("http.transport.protocol", "HTTP2");
+ peer = new Server(serverContext, Protocol.HTTP, 0, handler);
+ peer.start();
+ Context clientContext = new Context();
+ clientContext.getParameters().set("httpClientTransportMode", "HTTP2");
+ clientContext.getParameters().set("followRedirects", "false");
+ client = new Client(clientContext, Protocol.HTTP);
+ client.start();
+ adapter.setPlaintext(true);
+ adapter.setClient(client);
+ adapter.setTarget("localhost:" + peer.getActualPort());
+ adapter.start();
+ }
+
+ private Status failure() {
+ return assertThrows(
+ StatusRuntimeException.class,
+ () ->
+ ClientCalls.blockingUnaryCall(
+ adapter.getChannel(),
+ METHOD,
+ CallOptions.DEFAULT,
+ StringValue.of("request")))
+ .getStatus();
+ }
+
+ private static byte[] frame(String value) {
+ byte[] message = StringValue.of(value).toByteArray();
+ return ByteBuffer.allocate(5 + message.length)
+ .put((byte) 0)
+ .putInt(message.length)
+ .put(message)
+ .array();
+ }
+
+ private void startMalformedThenHealthy(Consumer malformed) throws Exception {
+ adapter.setMaxConcurrentCalls(1);
+ AtomicInteger requests = new AtomicInteger();
+ start(
+ new Restlet() {
+ @Override
+ public void handle(Request request, Response response) {
+ if (requests.incrementAndGet() == 1) {
+ malformed.accept(response);
+ } else {
+ response.setEntity(
+ new ByteArrayRepresentation(
+ frame("recovered"),
+ MediaType.valueOf("application/grpc")));
+ response.getTrailers().add("grpc-status", "0");
+ }
+ }
+ });
+ }
+
+ private void assertRecovered() {
+ assertEquals(
+ "recovered",
+ ClientCalls.blockingUnaryCall(
+ adapter.getChannel(),
+ METHOD,
+ CallOptions.DEFAULT,
+ StringValue.of("next"))
+ .getValue());
+ assertTrue(client.isStarted());
+ }
+
+ @AfterEach
+ void cleanup() throws Exception {
+ try {
+ adapter.stop();
+ } finally {
+ try {
+ if (client != null) {
+ client.stop();
+ }
+ } finally {
+ if (peer != null) {
+ peer.stop();
+ }
+ }
+ }
+ }
+
+ @Test
+ void sdkMapsHttpErrorsAndNoPostIsReplayed() throws Exception {
+ AtomicInteger requests = new AtomicInteger();
+ start(
+ new Restlet() {
+ @Override
+ public void handle(Request request, Response response) {
+ requests.incrementAndGet();
+ response.setStatus(org.restlet.data.Status.CLIENT_ERROR_UNAUTHORIZED);
+ response.setEntity("denied", MediaType.TEXT_PLAIN);
+ }
+ });
+ assertEquals(Status.Code.UNAUTHENTICATED, failure().getCode());
+ assertEquals(1, requests.get());
+ }
+
+ @Test
+ void missingGrpcStatusIsNotSuccess() throws Exception {
+ startMalformedThenHealthy(
+ response -> response.setEntity("", MediaType.valueOf("application/grpc")));
+ Status status = failure();
+ assertEquals(Status.Code.UNKNOWN, status.getCode(), status::toString);
+ assertTrue(status.getDescription().contains("missing GRPC status"));
+ assertRecovered();
+ }
+
+ @Test
+ void sdkRejectsTruncatedResponseFrame() throws Exception {
+ startMalformedThenHealthy(
+ response -> {
+ response.setEntity(
+ new ByteArrayRepresentation(
+ new byte[] {0, 0, 0, 0, 10, 1},
+ MediaType.valueOf("application/grpc")));
+ response.getTrailers().add("grpc-status", "0");
+ });
+ assertEquals(Status.Code.INTERNAL, failure().getCode());
+ assertRecovered();
+ }
+
+ @Test
+ void malformedBinaryMetadataFailsExplicitly() throws Exception {
+ startMalformedThenHealthy(
+ response -> {
+ response.setEntity("", MediaType.valueOf("application/grpc"));
+ response.getHeaders().add("x-invalid-bin", "!not base64!");
+ response.getTrailers().add("grpc-status", "0");
+ });
+ assertEquals(Status.Code.UNAVAILABLE, failure().getCode());
+ assertRecovered();
+ }
+
+ @Test
+ void wrongContentTypeFailsAndRecovers() throws Exception {
+ startMalformedThenHealthy(response -> response.setEntity("not grpc", MediaType.TEXT_PLAIN));
+ assertEquals(Status.Code.UNKNOWN, failure().getCode());
+ assertRecovered();
+ }
+
+ @Test
+ void invalidFrameFlagsFailAndRecover() throws Exception {
+ startMalformedThenHealthy(
+ response -> {
+ response.setEntity(
+ new ByteArrayRepresentation(
+ new byte[] {2, 0, 0, 0, 0},
+ MediaType.valueOf("application/grpc")));
+ response.getTrailers().add("grpc-status", "0");
+ });
+ assertEquals(Status.Code.INTERNAL, failure().getCode());
+ assertRecovered();
+ }
+
+ @Test
+ void multipleUnaryResponseMessagesFailAndRecover() throws Exception {
+ startMalformedThenHealthy(
+ response -> {
+ byte[] first = frame("one");
+ byte[] second = frame("two");
+ response.setEntity(
+ new ByteArrayRepresentation(
+ ByteBuffer.allocate(first.length + second.length)
+ .put(first)
+ .put(second)
+ .array(),
+ MediaType.valueOf("application/grpc")));
+ response.getTrailers().add("grpc-status", "0");
+ });
+ Status status = failure();
+ assertEquals(Status.Code.CANCELLED, status.getCode(), status::toString);
+ assertEquals(Status.Code.INTERNAL, Status.fromThrowable(status.getCause()).getCode());
+ assertTrue(
+ Status.fromThrowable(status.getCause())
+ .getDescription()
+ .contains("More than one value"));
+ assertRecovered();
+ }
+
+ @Test
+ void successfulStatusWithoutUnaryMessageFailsAndRecovers() throws Exception {
+ startMalformedThenHealthy(
+ response -> {
+ response.setEntity("", MediaType.valueOf("application/grpc"));
+ response.getTrailers().add("grpc-status", "0");
+ });
+ assertEquals(Status.Code.INTERNAL, failure().getCode());
+ assertRecovered();
+ }
+
+ @Test
+ void malformedStatusFailsAndRecovers() throws Exception {
+ startMalformedThenHealthy(
+ response -> {
+ response.setEntity("", MediaType.valueOf("application/grpc"));
+ response.getTrailers().add("grpc-status", "not-a-code");
+ });
+ assertEquals(Status.Code.UNKNOWN, failure().getCode());
+ assertRecovered();
+ }
+
+ @Test
+ void oversizedDeclaredMessageLengthFailsBeforePayloadAndRecovers() throws Exception {
+ adapter.setMaxMessageBytes(128);
+ startMalformedThenHealthy(
+ response -> {
+ response.setEntity(
+ new ByteArrayRepresentation(
+ ByteBuffer.allocate(5).put((byte) 0).putInt(129).array(),
+ MediaType.valueOf("application/grpc")));
+ response.getTrailers().add("grpc-status", "0");
+ });
+ assertEquals(Status.Code.RESOURCE_EXHAUSTED, failure().getCode());
+ assertRecovered();
+ }
+
+ @Test
+ void inboundWireBudgetRejectsOversizeResponseAndRecovers() throws Exception {
+ adapter.setMaxMessageBytes(128);
+ startMalformedThenHealthy(
+ response -> {
+ response.setEntity(
+ new ByteArrayRepresentation(
+ frame("x".repeat(127)), MediaType.valueOf("application/grpc")));
+ response.getTrailers().add("grpc-status", "0");
+ });
+ assertEquals(Status.Code.RESOURCE_EXHAUSTED, failure().getCode());
+ assertRecovered();
+ }
+}
diff --git a/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcModuleTestCase.java b/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcModuleTestCase.java
new file mode 100644
index 0000000000..1133978ae6
--- /dev/null
+++ b/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcModuleTestCase.java
@@ -0,0 +1,142 @@
+/**
+ * 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 java.io.File;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.restlet.Client;
+
+class GrpcModuleTestCase {
+ @Test
+ void nativeChannelResolvesWithoutSdkNetworkTransportOrInternalExports(@TempDir Path temp)
+ throws Exception {
+ String classPath =
+ System.getProperty(
+ "surefire.test.class.path", System.getProperty("java.class.path"));
+ var paths =
+ new ArrayList<>(
+ Arrays.stream(classPath.split(File.pathSeparator))
+ .map(Path::of)
+ .filter(Files::isRegularFile)
+ .filter(path -> path.toString().endsWith(".jar"))
+ .filter(
+ path ->
+ !path.getFileName()
+ .toString()
+ .startsWith("grpc-netty"))
+ .toList());
+ paths.add(
+ Path.of(
+ System.getProperty(
+ "restlet.consumer.core",
+ Path.of(
+ Client.class
+ .getProtectionDomain()
+ .getCodeSource()
+ .getLocation()
+ .toURI())
+ .toString())));
+ paths.add(
+ Path.of(
+ System.getProperty(
+ "restlet.consumer.grpc",
+ Path.of(
+ GrpcClientAdapter.class
+ .getProtectionDomain()
+ .getCodeSource()
+ .getLocation()
+ .toURI())
+ .toString())));
+ String modulePath =
+ paths.stream()
+ .distinct()
+ .map(Path::toString)
+ .collect(Collectors.joining(File.pathSeparator));
+ Path descriptor = temp.resolve("module-info.java");
+ Path example = temp.resolve("Main.java");
+ Files.writeString(descriptor, "module r3.consumer { requires org.restlet.ext.grpc; }");
+ Files.writeString(
+ example,
+ """
+ package example;
+ import org.restlet.Client;
+ import org.restlet.Context;
+ import org.restlet.data.Protocol;
+ import org.restlet.ext.grpc.GrpcClientAdapter;
+ public final class Main {
+ public static void main(String[] args) throws Exception {
+ try {
+ Class.forName("io.grpc.netty.shaded.io.grpc.netty.NettyChannelBuilder");
+ throw new AssertionError("SDK network transport is present");
+ } catch (ClassNotFoundException expected) { }
+ var context = new Context();
+ context.getParameters().set("httpClientTransportMode", "HTTP2");
+ context.getParameters().set("followRedirects", "false");
+ var client = new Client(context, Protocol.HTTP);
+ client.start();
+ try (var adapter = new GrpcClientAdapter()) {
+ adapter.setClient(client);
+ adapter.setPlaintext(true);
+ adapter.setTarget("dns:///localhost:12345");
+ adapter.start();
+ if (!adapter.getChannel().authority().equals("localhost:12345")) {
+ throw new AssertionError("Native authority was lost");
+ }
+ } finally { client.stop(); }
+ }
+ }
+ """);
+ Path classes = temp.resolve("classes");
+ var diagnostics = new java.io.ByteArrayOutputStream();
+ assertEquals(
+ 0,
+ javax.tools.ToolProvider.getSystemJavaCompiler()
+ .run(
+ null,
+ diagnostics,
+ diagnostics,
+ "--module-path",
+ modulePath,
+ "-d",
+ classes.toString(),
+ descriptor.toString(),
+ example.toString()),
+ diagnostics.toString());
+ Path output = temp.resolve("consumer.log");
+ Process child =
+ new ProcessBuilder(
+ Path.of(System.getProperty("java.home"), "bin", "java").toString(),
+ "--module-path",
+ modulePath + File.pathSeparator + classes,
+ "--module",
+ "r3.consumer/example.Main")
+ .redirectErrorStream(true)
+ .redirectOutput(output.toFile())
+ .start();
+ try {
+ assertTrue(child.waitFor(30, TimeUnit.SECONDS), "Named consumer did not terminate");
+ assertEquals(0, child.exitValue(), Files.readString(output));
+ } finally {
+ if (child.isAlive()) {
+ child.destroyForcibly();
+ assertTrue(child.waitFor(10, TimeUnit.SECONDS));
+ }
+ }
+ }
+}
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
new file mode 100644
index 0000000000..a51d69e326
--- /dev/null
+++ b/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/internal/RestletClientStreamTestCase.java
@@ -0,0 +1,100 @@
+/**
+ * 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.internal;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertSame;
+
+import io.grpc.CallOptions;
+import io.grpc.ClientStreamTracer;
+import io.grpc.Metadata;
+import io.grpc.MethodDescriptor;
+import io.grpc.Status;
+import io.grpc.internal.ClientStreamListener;
+import io.grpc.internal.StreamListener.MessageProducer;
+import java.io.ByteArrayInputStream;
+import java.io.InputStream;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import org.junit.jupiter.api.Test;
+import org.restlet.Client;
+import org.restlet.data.Protocol;
+
+class RestletClientStreamTestCase {
+ @Test
+ void shutdownBeforeListenerInstallationIsReplayedOnceAndReleasesAdmission() {
+ var marshaller =
+ new MethodDescriptor.Marshaller() {
+ @Override
+ public InputStream stream(byte[] value) {
+ return new ByteArrayInputStream(value);
+ }
+
+ @Override
+ public byte[] parse(InputStream input) {
+ throw new UnsupportedOperationException("No response is expected");
+ }
+ };
+ var method =
+ MethodDescriptor.newBuilder()
+ .setType(MethodDescriptor.MethodType.UNARY)
+ .setFullMethodName("review.Unary/Echo")
+ .setRequestMarshaller(marshaller)
+ .setResponseMarshaller(marshaller)
+ .build();
+ AtomicInteger released = new AtomicInteger();
+ AtomicInteger terminals = new AtomicInteger();
+ AtomicReference status = new AtomicReference<>();
+ try (var io = Executors.newVirtualThreadPerTaskExecutor()) {
+ var stream =
+ new RestletClientStream(
+ new Client(Protocol.HTTP),
+ new InetSocketAddress(InetAddress.getLoopbackAddress(), 8080),
+ "localhost:8080",
+ method,
+ new Metadata(),
+ CallOptions.DEFAULT,
+ new ClientStreamTracer[0],
+ 128,
+ io,
+ ignored -> released.incrementAndGet());
+ Status shutdown = Status.UNAVAILABLE.withDescription("shutdown before start");
+ stream.cancel(shutdown);
+ stream.cancel(Status.CANCELLED);
+ assertEquals(0, terminals.get());
+ assertEquals(0, released.get());
+ new GuardedClientStream(stream, 128)
+ .start(
+ new ClientStreamListener() {
+ @Override
+ public void headersRead(Metadata headers) {}
+
+ @Override
+ public void closed(
+ Status value, RpcProgress progress, Metadata trailers) {
+ status.set(value);
+ terminals.incrementAndGet();
+ }
+
+ @Override
+ public void messagesAvailable(MessageProducer producer) {}
+
+ @Override
+ public void onReady() {}
+ });
+ stream.cancel(Status.CANCELLED);
+ assertSame(shutdown, status.get());
+ assertEquals(1, terminals.get());
+ assertEquals(1, released.get());
+ }
+ }
+}
diff --git a/pom.xml b/pom.xml
index 8170543aba..eb7f2c4122 100644
--- a/pom.xml
+++ b/pom.xml
@@ -53,6 +53,7 @@
2.3.35
2.14.0
+ 1.85.1
2.22.3
3.2.3
2.22
@@ -77,6 +78,7 @@
org.restlet.ext.crypto
org.restlet.ext.freemarker
org.restlet.ext.gson
+ org.restlet.ext.grpc
org.restlet.ext.jaas
org.restlet.ext.jackson
org.restlet.ext.json
From 62118b8bffbd174e6e2f6f791717dcb9abc7bf14 Mon Sep 17 00:00:00 2001
From: Jerome Louvel <374450+jlouvel@users.noreply.github.com>
Date: Sat, 31 Oct 2026 02:30:16 -0400
Subject: [PATCH 2/4] fix: abort the gRPC HTTP exchange when deframing fails
A deframing failure, such as an unsupported grpc-encoding, now aborts the
underlying HTTP exchange so that its worker does not stay blocked reading
an unfinished response body.
Co-Authored-By: Claude Opus 5.5
---
.../grpc/internal/RestletClientStream.java | 3 +-
.../ext/grpc/GrpcHttpValidationTestCase.java | 36 +++++++++++++++++++
2 files changed, 38 insertions(+), 1 deletion(-)
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 67edd46a96..d22ab4e941 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
@@ -368,7 +368,8 @@ public void bytesRead(int count) {
@Override
public void deframeFailed(Throwable cause) {
- fail(Status.fromThrowable(cause));
+ // Also abort the HTTP exchange, whose worker may be blocked reading the body.
+ RestletClientStream.this.fail(Status.fromThrowable(cause));
}
@Override
diff --git a/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcHttpValidationTestCase.java b/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcHttpValidationTestCase.java
index c285658103..8d1f1d7f96 100644
--- a/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcHttpValidationTestCase.java
+++ b/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcHttpValidationTestCase.java
@@ -19,7 +19,11 @@
import io.grpc.StatusRuntimeException;
import io.grpc.protobuf.ProtoUtils;
import io.grpc.stub.ClientCalls;
+import java.io.IOException;
+import java.io.OutputStream;
import java.nio.ByteBuffer;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
import org.junit.jupiter.api.AfterEach;
@@ -33,6 +37,7 @@
import org.restlet.data.MediaType;
import org.restlet.data.Protocol;
import org.restlet.representation.ByteArrayRepresentation;
+import org.restlet.representation.OutputRepresentation;
class GrpcHttpValidationTestCase {
private final GrpcClientAdapter adapter = new GrpcClientAdapter();
@@ -280,4 +285,35 @@ void inboundWireBudgetRejectsOversizeResponseAndRecovers() throws Exception {
assertEquals(Status.Code.RESOURCE_EXHAUSTED, failure().getCode());
assertRecovered();
}
+
+ @Test
+ void unsupportedEncodingAbortsAnUnfinishedResponseAndRecovers() throws Exception {
+ CountDownLatch aborted = new CountDownLatch(1);
+ startMalformedThenHealthy(
+ response -> {
+ response.getHeaders().add("grpc-encoding", "unsupported-encoding");
+ response.setEntity(
+ new OutputRepresentation(MediaType.valueOf("application/grpc")) {
+ @Override
+ public void write(OutputStream output) throws IOException {
+ try {
+ for (int i = 0; i < 100; i++) {
+ output.write(0);
+ output.flush();
+ Thread.sleep(100);
+ }
+ } catch (IOException e) {
+ aborted.countDown();
+ throw e;
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IOException(e);
+ }
+ }
+ });
+ });
+ assertEquals(Status.Code.INTERNAL, failure().getCode());
+ assertTrue(aborted.await(5, TimeUnit.SECONDS), "The HTTP exchange must be aborted");
+ assertRecovered();
+ }
}
From 8464ff7561839005f71898ff88ed457060300771 Mon Sep 17 00:00:00 2001
From: Jerome Louvel <374450+jlouvel@users.noreply.github.com>
Date: Sat, 31 Oct 2026 02:30:16 -0400
Subject: [PATCH 3/4] fix: remove cancelled gRPC deadline timers from the
scheduler queue
Co-Authored-By: Claude Opus 5.5
---
.../grpc/internal/RestletTransportFactory.java | 9 ++++++---
.../internal/RestletClientStreamTestCase.java | 16 ++++++++++++++++
2 files changed, 22 insertions(+), 3 deletions(-)
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 19578d1d2e..dd28450b53 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
@@ -34,6 +34,7 @@
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import org.restlet.Client;
@@ -46,9 +47,9 @@ public final class RestletTransportFactory implements ClientTransportFactory {
private final Semaphore admission;
private final Set transports = new HashSet<>();
private final ExecutorService io = Executors.newVirtualThreadPerTaskExecutor();
- private final ScheduledExecutorService timer =
- Executors.newSingleThreadScheduledExecutor(
- Thread.ofPlatform().daemon().name("restlet-grpc-timer").factory());
+ private final ScheduledThreadPoolExecutor timer =
+ new ScheduledThreadPoolExecutor(
+ 1, Thread.ofPlatform().daemon().name("restlet-grpc-timer").factory());
private boolean closed;
public RestletTransportFactory(Client client, int maxBytes, int maxCalls, Duration timeout) {
@@ -56,6 +57,8 @@ public RestletTransportFactory(Client client, int maxBytes, int maxCalls, Durati
this.maxBytes = maxBytes;
admission = new Semaphore(maxCalls);
shutdownTimeout = timeout;
+ // Cancelled deadline timers of completed calls must not stay queued until they expire.
+ timer.setRemoveOnCancelPolicy(true);
}
@Override
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 a51d69e326..41bfd05f82 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
@@ -23,6 +23,8 @@
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import org.junit.jupiter.api.Test;
@@ -30,6 +32,20 @@
import org.restlet.data.Protocol;
class RestletClientStreamTestCase {
+ @Test
+ void cancelledTimersDoNotStayQueued() {
+ var factory = new RestletTransportFactory(null, 1024, 1, java.time.Duration.ofSeconds(1));
+ try {
+ var timer = (ScheduledThreadPoolExecutor) factory.getScheduledExecutorService();
+ for (int i = 0; i < 100; i++) {
+ timer.schedule(() -> {}, 1, TimeUnit.HOURS).cancel(false);
+ }
+ assertEquals(0, timer.getQueue().size());
+ } finally {
+ factory.close();
+ }
+ }
+
@Test
void shutdownBeforeListenerInstallationIsReplayedOnceAndReleasesAdmission() {
var marshaller =
From ef684476ad287b01cc61a29ad7d495a270fbdb74 Mon Sep 17 00:00:00 2001
From: Jerome Louvel <374450+jlouvel@users.noreply.github.com>
Date: Sat, 31 Oct 2026 02:30:17 -0400
Subject: [PATCH 4/4] fix: apply the gRPC call duration cap after application
interceptors
Register the cap first so that it runs last and clamps the call options
passed to the transport, even if an interceptor sets a longer deadline.
Co-Authored-By: Claude Opus 5.5
---
.../restlet/ext/grpc/GrpcClientAdapter.java | 6 ++++--
.../ext/grpc/GrpcClientAdapterTestCase.java | 21 +++++++++++++++++++
2 files changed, 25 insertions(+), 2 deletions(-)
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 5a1c208aaf..3847e932e0 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
@@ -194,7 +194,7 @@ public synchronized void start() {
.disableServiceConfigLookUp()
.proxyDetector(address -> null)
.maxInboundMessageSize(maxMessageBytes)
- .intercept(List.copyOf(interceptors))
+ // Registered first so that it clamps the final call options.
.intercept(
new ClientInterceptor() {
@Override
@@ -214,7 +214,9 @@ public ClientCall interceptCall(
? cap
: deadline.minimum(cap)));
}
- });
+ })
+ // Registered last so they run first, before the duration cap.
+ .intercept(List.copyOf(interceptors));
if (executor != null) {
builder.executor(executor);
}
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 38f53b321e..eb15d91ec9 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
@@ -613,6 +613,27 @@ void enforcesDefaultCallDurationAndCallerMayShortenIt() throws Exception {
assertEquals("next", call("next").getValue());
}
+ @Test
+ void interceptorCannotExtendAdapterDuration() throws Exception {
+ adapter.setMaxCallDuration(Duration.ofSeconds(3));
+ adapter.getInterceptors()
+ .add(
+ new io.grpc.ClientInterceptor() {
+ @Override
+ public ClientCall interceptCall(
+ MethodDescriptor method,
+ CallOptions options,
+ io.grpc.Channel next) {
+ return next.newCall(
+ method, options.withDeadlineAfter(1, TimeUnit.HOURS));
+ }
+ });
+ Pending pending = startBlocked();
+ var failure = assertThrows(StatusRuntimeException.class, () -> call("block"));
+ assertEquals(Status.Code.DEADLINE_EXCEEDED, failure.getStatus().getCode());
+ assertTrue(pending.cancelled().await(30, TimeUnit.SECONDS));
+ }
+
@Test
void callerDeadlineCanShortenAdapterDuration() throws Exception {
adapter.setMaxCallDuration(Duration.ofSeconds(20));