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..3847e932e0
--- /dev/null
+++ b/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/GrpcClientAdapter.java
@@ -0,0 +1,292 @@
+/**
+ * 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)
+ // Registered first so that it clamps the final call options.
+ .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)));
+ }
+ })
+ // Registered last so they run first, before the duration cap.
+ .intercept(List.copyOf(interceptors));
+ 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..d22ab4e941
--- /dev/null
+++ b/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletClientStream.java
@@ -0,0 +1,428 @@
+/**
+ * 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) {
+ // Also abort the HTTP exchange, whose worker may be blocked reading the body.
+ RestletClientStream.this.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..dd28450b53
--- /dev/null
+++ b/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletTransportFactory.java
@@ -0,0 +1,265 @@
+/**
+ * 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.ScheduledThreadPoolExecutor;
+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 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) {
+ this.client = client;
+ 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
+ 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..eb15d91ec9
--- /dev/null
+++ b/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcClientAdapterTestCase.java
@@ -0,0 +1,780 @@
+/**
+ * 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 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));
+ 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..8d1f1d7f96
--- /dev/null
+++ b/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcHttpValidationTestCase.java
@@ -0,0 +1,319 @@
+/**
+ * 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.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;
+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;
+import org.restlet.representation.OutputRepresentation;
+
+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();
+ }
+
+ @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();
+ }
+}
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..41bfd05f82
--- /dev/null
+++ b/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/internal/RestletClientStreamTestCase.java
@@ -0,0 +1,116 @@
+/**
+ * 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.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+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 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 =
+ 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