diff --git a/changes.md b/changes.md index 9911f0ced7..358c09c631 100644 --- a/changes.md +++ b/changes.md @@ -95,6 +95,15 @@ Changes log bounded messages/calls, one-shot SSE responses, and request cancellation. Sessions, listening GET streams, replay and Ikanos revision compatibility are not supported. Borrowed HTTP connectors are not stopped by the MCP client. + - Added experimental SDK-native gRPC unary and bounded server-streaming hosting on a + borrowed Restlet/Netty HTTP/2 listener. Generated and dynamic services can share the + listener with REST routes. grpc-java owns service dispatch, framing and status semantics; + Restlet owns HTTP I/O and finite message, queue, concurrency and call-duration budgets. + Cancellation and response completion terminate each logical transport exactly once; + admission remains held until response transmission or cancellation. Adapter shutdown + leaves the borrowed listener and supplied callback executor running and reports timeout + explicitly. Client-streaming, bidirectional streaming, TLS configuration and physical + gRPC connection controls are not supported by the server adapter. - Added an experimental SDK-native gRPC unary client on a borrowed Restlet HTTP/2 connector, with explicit plaintext opt-in, bounded messages and call admission. Regression tests cover authority mapping, cancellation before and after headers, diff --git a/org.restlet.ext.grpc/pom.xml b/org.restlet.ext.grpc/pom.xml index 8e4f53f1f1..61a62faf25 100644 --- a/org.restlet.ext.grpc/pom.xml +++ b/org.restlet.ext.grpc/pom.xml @@ -9,9 +9,14 @@ org.restlet.ext.grpc Restlet Framework - gRPC extension - Native gRPC unary client on a borrowed Restlet HTTP/2 connector. + Native gRPC unary client and unary/server-streaming hosting on borrowed Restlet HTTP/2 connectors. + + com.google.guava + failureaccess + 1.0.3 + com.google.guava guava diff --git a/org.restlet.ext.grpc/src/main/java/module-info.java b/org.restlet.ext.grpc/src/main/java/module-info.java index e5da89d47b..c3092716f8 100644 --- a/org.restlet.ext.grpc/src/main/java/module-info.java +++ b/org.restlet.ext.grpc/src/main/java/module-info.java @@ -12,6 +12,8 @@ requires transitive io.grpc; requires io.grpc.internal; requires com.google.common; + // Guava is automatic and does not resolve this explicit module itself. + requires com.google.common.util.concurrent.internal; exports org.restlet.ext.grpc; } diff --git a/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/GrpcServerAdapter.java b/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/GrpcServerAdapter.java new file mode 100644 index 0000000000..8f698ed6dd --- /dev/null +++ b/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/GrpcServerAdapter.java @@ -0,0 +1,314 @@ +/** + * 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.ServerInterceptor; +import io.grpc.ServerServiceDefinition; +import io.grpc.internal.ServerImplBuilder; +import java.io.IOException; +import java.time.Duration; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import org.restlet.Context; +import org.restlet.Request; +import org.restlet.Response; +import org.restlet.Restlet; +import org.restlet.ext.grpc.internal.RestletServerHost; +import org.restlet.routing.Router; +import org.restlet.routing.Template; + +/** + * Socketless grpc-java server on a borrowed Restlet HTTP/2 listener. + * + *

Supports unary requests and server-streamed responses, with finite message, queue, concurrency + * and duration budgets. Does not implement TLS or physical connection controls. The host + * application owns the HTTP listener and any supplied callback executor. + * + *

Register services before {@link #start()}, attach them to a router with {@link #attachTo}, and + * configure the host listener for plaintext HTTP/2. This adapter is single-use. Stopping it cancels + * active calls, waits up to the shutdown timeout for SDK and owned callback termination, and shuts + * down only its own executor and timer, not the listener or an executor supplied with {@link + * #setExecutor}. + */ +public final class GrpcServerAdapter extends Restlet { + private final List services = new CopyOnWriteArrayList<>(); + private final List interceptors = new CopyOnWriteArrayList<>(); + + /** + * Guards the configuration and lifecycle state. The adapter's own monitor, held by {@link + * #start()} and {@link #stop()} like any Restlet's, is taken first. + */ + private final Object lock = new Object(); + + private int maxMessageBytes = 1024 * 1024; + private int maxQueuedBytes = 256 * 1024; + private int maxConcurrentCalls = 128; + private Duration maxCallDuration = Duration.ofSeconds(5); + private Duration shutdownTimeout = Duration.ofSeconds(5); + private Executor executor; + private ExecutorService ownedExecutor; + private io.grpc.Server sdk; + private RestletServerHost host; + private boolean configured; + + public GrpcServerAdapter() { + this(new Context()); + } + + public GrpcServerAdapter(Context context) { + super(context); + } + + public List getServices() { + return services; + } + + public List getInterceptors() { + return interceptors; + } + + public void setMaxMessageBytes(int value) { + synchronized (lock) { + configurable(); + if (value <= 0 || value > Integer.MAX_VALUE - 6) { + throw new IllegalArgumentException("A positive finite message budget is required"); + } + maxMessageBytes = value; + } + } + + public int getMaxMessageBytes() { + synchronized (lock) { + return maxMessageBytes; + } + } + + public void setMaxQueuedBytes(int value) { + synchronized (lock) { + configurable(); + if (value < 16384) { + throw new IllegalArgumentException("Queue budget must be at least 16384 bytes"); + } + maxQueuedBytes = value; + } + } + + public int getMaxQueuedBytes() { + synchronized (lock) { + return maxQueuedBytes; + } + } + + public void setMaxConcurrentCalls(int value) { + synchronized (lock) { + configurable(); + if (value <= 0) { + throw new IllegalArgumentException("A positive call budget is required"); + } + maxConcurrentCalls = value; + } + } + + public int getMaxConcurrentCalls() { + synchronized (lock) { + return maxConcurrentCalls; + } + } + + public void setMaxCallDuration(Duration value) { + synchronized (lock) { + configurable(); + maxCallDuration = positive(value); + } + } + + public Duration getMaxCallDuration() { + synchronized (lock) { + return maxCallDuration; + } + } + + public void setShutdownTimeout(Duration value) { + synchronized (lock) { + configurable(); + shutdownTimeout = positive(value); + } + } + + public Duration getShutdownTimeout() { + synchronized (lock) { + return shutdownTimeout; + } + } + + public void setExecutor(Executor value) { + synchronized (lock) { + configurable(); + executor = java.util.Objects.requireNonNull(value, "executor"); + } + } + + /** + * Returns the callback executor: the supplied one, the adapter-owned one once started, or null + * before start when none was supplied. + */ + public Executor getExecutor() { + synchronized (lock) { + return executor; + } + } + + private static Duration positive(Duration value) { + if (value.isNegative() || value.isZero()) { + throw new IllegalArgumentException("A positive duration is required"); + } + return Duration.ofNanos(value.toNanos()); + } + + /** Requires the configuration to be still open; called with {@link #lock} held. */ + private void configurable() { + if (configured) { + throw new IllegalStateException("Server adapter is single-use; configuration is fixed"); + } + } + + /** Attaches gRPC service prefixes while leaving other routes on the same listener untouched. */ + public void attachTo(Router router) { + for (var service : List.copyOf(services)) { + router.attach( + "/" + service.getServiceDescriptor().getName() + "/", + this, + Template.MODE_STARTS_WITH); + } + } + + @Override + public synchronized void start() throws Exception { + synchronized (lock) { + if (isStarted()) { + return; + } + configurable(); + for (var service : List.copyOf(services)) { + for (var method : service.getMethods()) { + var type = method.getMethodDescriptor().getType(); + if (type != io.grpc.MethodDescriptor.MethodType.UNARY + && type != io.grpc.MethodDescriptor.MethodType.SERVER_STREAMING) { + throw new UnsupportedOperationException( + "Only unary and server-streaming methods are supported"); + } + } + } + configured = true; + host = + new RestletServerHost( + maxMessageBytes, maxQueuedBytes, maxConcurrentCalls, maxCallDuration); + if (executor == null) { + ownedExecutor = Executors.newVirtualThreadPerTaskExecutor(); + executor = ownedExecutor; + } + var builder = + new ServerImplBuilder( + (tracers, recorder) -> { + host.setTracers(tracers); + return host; + }) + .executor(executor) + .maxInboundMessageSize(maxMessageBytes); + List.copyOf(services).forEach(builder::addService); + List.copyOf(interceptors).forEach(builder::intercept); + // Added last, so outermost: it holds a call's callback slot around all of them. + builder.intercept(host.callbackBudget()); + try { + sdk = builder.build().start(); + super.start(); + } catch (IOException | RuntimeException e) { + host.close(); + if (ownedExecutor != null) { + ownedExecutor.shutdownNow(); + } + throw e; + } + } + } + + @Override + public void handle(Request request, Response response) { + RestletServerHost current; + synchronized (lock) { + current = isStarted() ? host : null; + } + if (current == null) { + response.setStatus(org.restlet.data.Status.SERVER_ERROR_SERVICE_UNAVAILABLE); + return; + } + current.handle(request, response); + } + + /** + * Stops the adapter. The waits only hold the adapter's monitor, not {@link #lock}, so callbacks + * finishing meanwhile may still use its accessors. + */ + @Override + public synchronized void stop() throws Exception { + io.grpc.Server server; + RestletServerHost stopping; + ExecutorService owned; + Duration timeout; + synchronized (lock) { + if (sdk == null) { + return; + } + server = sdk; + stopping = host; + owned = ownedExecutor; + timeout = shutdownTimeout; + sdk = null; + } + long started = System.nanoTime(); + // Cancelling calls may run listeners, so the timeout bounds it like the waits after it. + Thread cancelling = + Thread.ofVirtual() + .name("restlet-grpc-server-stop") + .start( + () -> { + server.shutdownNow(); + if (stopping != null) { + stopping.close(); + } + }); + try { + if (!cancelling.join(timeout) + || !server.awaitTermination( + Math.max(0, timeout.toNanos() - (System.nanoTime() - started)), + TimeUnit.NANOSECONDS)) { + throw new IllegalStateException("gRPC server shutdown timed out"); + } + if (owned != null) { + owned.shutdown(); + long remaining = timeout.toNanos() - (System.nanoTime() - started); + if (!owned.awaitTermination(Math.max(0, remaining), TimeUnit.NANOSECONDS)) { + throw new IllegalStateException("gRPC callback executor shutdown timed out"); + } + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw e; + } finally { + if (owned != null) { + owned.shutdownNow(); + } + super.stop(); + } + } +} diff --git a/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletServerHost.java b/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletServerHost.java new file mode 100644 index 0000000000..21a5af88ca --- /dev/null +++ b/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletServerHost.java @@ -0,0 +1,845 @@ +/** + * 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.ForwardingServerCallListener; +import io.grpc.Grpc; +import io.grpc.InternalChannelz; +import io.grpc.InternalInstrumented; +import io.grpc.InternalLogId; +import io.grpc.InternalMetadata; +import io.grpc.InternalStatus; +import io.grpc.Metadata; +import io.grpc.ServerCall; +import io.grpc.ServerCallHandler; +import io.grpc.ServerInterceptor; +import io.grpc.ServerStreamTracer; +import io.grpc.Status; +import io.grpc.internal.AbstractServerStream; +import io.grpc.internal.InternalServer; +import io.grpc.internal.ReadableBuffers; +import io.grpc.internal.SerializingExecutor; +import io.grpc.internal.ServerListener; +import io.grpc.internal.ServerTransport; +import io.grpc.internal.ServerTransportListener; +import io.grpc.internal.StatsTraceContext; +import io.grpc.internal.TransportFrameUtil; +import io.grpc.internal.TransportTracer; +import io.grpc.internal.WritableBuffer; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.SocketAddress; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.List; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.ScheduledThreadPoolExecutor; +import java.util.concurrent.Semaphore; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; +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.Protocol; +import org.restlet.engine.header.HeaderConstants; +import org.restlet.representation.OutputRepresentation; +import org.restlet.util.Series; + +/** Pinned SDK socketless server transport; each transport represents one logical HTTP exchange. */ +@SuppressWarnings("null") +public final class RestletServerHost implements InternalServer, AutoCloseable { + /** How long a cancelled call may stay blocked in one network read or write before a reset. */ + private static final Duration STUCK_IO = Duration.ofMillis(200); + + /** The logical call of an SDK stream, among its attributes. */ + private static final Attributes.Key CALL = Attributes.Key.create("restlet-grpc-call"); + + /** Metadata that would break the gRPC framing if mapped onto the HTTP response. */ + private static final Set FRAMING_HEADERS = + Set.of( + "content-type", + "content-length", + "content-encoding", + "transfer-encoding", + "connection"); + + private final int maxBytes; + private final int queueBytes; + private final Duration duration; + private final Semaphore admission; + + /** + * Calls whose callbacks may still run. Cancelled or completed calls leave admission at once but + * keep one of these until their callbacks exit, up to as many again as admitted calls. + */ + private final Semaphore callbacks; + + private final Set calls = ConcurrentHashMap.newKeySet(); + private final ScheduledThreadPoolExecutor timer = + new ScheduledThreadPoolExecutor( + 1, Thread.ofPlatform().daemon().name("restlet-grpc-server-timer").factory()); + + /** + * Runs the scheduled work of the SDK, such as client deadlines whose cancellation listeners may + * block, apart from the call budgets of {@link #timer}. + */ + private final ScheduledThreadPoolExecutor sdkTimer = + new ScheduledThreadPoolExecutor( + 1, Thread.ofPlatform().daemon().name("restlet-grpc-sdk-timer").factory()); + + private volatile ServerListener listener; + private List tracers = List.of(); + private boolean closed; + + public RestletServerHost(int maxBytes, int queueBytes, int maxCalls, Duration duration) { + this.maxBytes = maxBytes; + this.queueBytes = queueBytes; + this.duration = duration; + admission = new Semaphore(maxCalls); + callbacks = new Semaphore(maxCalls > Integer.MAX_VALUE / 2 ? maxCalls : maxCalls * 2); + // Each completed call cancels its deadline, which must not stay queued until it expires. + timer.setRemoveOnCancelPolicy(true); + sdkTimer.setRemoveOnCancelPolicy(true); + } + + public void setTracers(List value) { + tracers = List.copyOf(value); + } + + @Override + public synchronized void start(ServerListener value) { + listener = value; + } + + @Override + public synchronized void shutdown() { + if (!closed) { + closed = true; + for (Call call : List.copyOf(calls)) { + call.cancel(Status.UNAVAILABLE.withDescription("Server shutdown")); + } + if (listener != null) { + listener.serverShutdown(); + } + } + } + + @Override + public void close() { + shutdown(); + calls.forEach(call -> call.cancel(Status.UNAVAILABLE.withDescription("Adapter stopped"))); + timer.shutdownNow(); + sdkTimer.shutdownNow(); + } + + @Override + public SocketAddress getListenSocketAddress() { + return new InetSocketAddress("localhost", 0); + } + + @Override + public List getListenSocketAddresses() { + return List.of(); + } + + @Override + public InternalInstrumented getListenSocketStats() { + return null; + } + + @Override + public List> getListenSocketStatsList() { + return List.of(); + } + + public void handle(Request request, Response response) { + if (!Method.POST.equals(request.getMethod())) { + response.setStatus(org.restlet.data.Status.CLIENT_ERROR_METHOD_NOT_ALLOWED); + return; + } + if (!isGrpc(request.getEntity() == null ? null : request.getEntity().getMediaType())) { + response.setStatus(org.restlet.data.Status.CLIENT_ERROR_UNSUPPORTED_MEDIA_TYPE); + return; + } + String method = method(request); + if (method == null) { + response.setStatus(org.restlet.data.Status.CLIENT_ERROR_NOT_FOUND); + return; + } + // Over HTTP/1.1, Netty decides on trailers before the entity can add grpc-status. + Protocol protocol = request.getProtocol(); + if (protocol == null || !"2.0".equals(protocol.getVersion())) { + response.setStatus(org.restlet.data.Status.SERVER_ERROR_VERSION_NOT_SUPPORTED); + return; + } + Call call; + synchronized (this) { + if (closed) { + response.setStatus(org.restlet.data.Status.SERVER_ERROR_SERVICE_UNAVAILABLE); + return; + } + if (!admission.tryAcquire()) { + response.setStatus(org.restlet.data.Status.SERVER_ERROR_SERVICE_UNAVAILABLE); + return; + } + if (!callbacks.tryAcquire()) { + admission.release(); + response.setStatus(org.restlet.data.Status.SERVER_ERROR_SERVICE_UNAVAILABLE); + return; + } + call = new Call(request, response, method); + calls.add(call); + call.transportListener = listener.transportCreated(call); + } + try { + call.begin(); + } catch (IOException | RuntimeException e) { + call.cancel( + Status.INVALID_ARGUMENT.withDescription("Invalid gRPC request").withCause(e)); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + call.cancel(Status.CANCELLED.withCause(e)); + } + } + + /** + * Returns the interceptor holding a call's callback slot until the SDK reports its end to the + * application, after its earlier callbacks exited. It must be the outermost interceptor. + */ + public ServerInterceptor callbackBudget() { + return new ServerInterceptor() { + @Override + public ServerCall.Listener interceptCall( + ServerCall call, Metadata headers, ServerCallHandler next) { + Call owner = call.getAttributes().get(CALL); + if (owner == null) { + return next.startCall(call, headers); + } + if (!owner.callbacksStarted()) { + // The call ended and released its slot before its callbacks could start. + call.close( + Status.CANCELLED.withDescription("gRPC call ended before it started"), + new Metadata()); + return new ServerCall.Listener<>() {}; + } + ServerCall.Listener listener; + try { + listener = next.startCall(call, headers); + } catch (RuntimeException | Error e) { + owner.callbacksEnded(); + throw e; + } + return new ForwardingServerCallListener.SimpleForwardingServerCallListener<>( + listener) { + @Override + public void onComplete() { + try { + super.onComplete(); + } finally { + owner.callbacksEnded(); + } + } + + @Override + public void onCancel() { + try { + super.onCancel(); + } finally { + owner.callbacksEnded(); + } + } + }; + } + }; + } + + /** Accepts application/grpc and application/grpc+*, whatever their parameters. */ + private static boolean isGrpc(MediaType type) { + return type != null + && "application".equalsIgnoreCase(type.getMainType()) + && type.getSubType() != null + && (type.getSubType().equalsIgnoreCase("grpc") + || type.getSubType() + .toLowerCase(java.util.Locale.ROOT) + .startsWith("grpc+")); + } + + /** + * Returns the gRPC method of the raw path that the router matched, or null unless it is exactly + * "/service/method" without dot segments: a path normalized differently than routed could + * otherwise reach a method outside its route. + */ + private static String method(Request request) { + String path = request.getResourceRef() == null ? null : request.getResourceRef().getPath(); + if (path == null || !path.startsWith("/")) { + return null; + } + String[] segments = path.substring(1).split("/", -1); + if (segments.length != 2) { + return null; + } + for (String segment : segments) { + if (segment.isEmpty() || ".".equals(segment) || "..".equals(segment)) { + return null; + } + } + return path.substring(1); + } + + private final class Call implements ServerTransport { + private final Request request; + private final Response response; + private final String method; + + /** + * Runs SDK notifications, and application callbacks with a direct executor, in virtual + * threads: they never block the HTTP dispatch, nor its cancellation. + */ + private final SerializingExecutor serial = + new SerializingExecutor( + task -> Thread.ofVirtual().name("restlet-grpc-server-call").start(task)); + + private final InternalLogId id = InternalLogId.allocate(Call.class, "Restlet logical call"); + private final ArrayDeque queue = new ArrayDeque<>(); + private final CountDownLatch head = new CountDownLatch(1); + private final AtomicBoolean finished = new AtomicBoolean(); + private final AtomicLong operations = new AtomicLong(); + + /** 0 before the application callbacks, 1 while they may run, 2 once released. */ + private final AtomicInteger callbackState = new AtomicInteger(); + + /** The network read or write in progress, 0 if none. */ + private final AtomicLong blocked = new AtomicLong(); + + private ServerTransportListener transportListener; + private SdkStream stream; + private volatile ScheduledFuture deadline; + private Metadata trailers; + private int queued; + + /** Bytes framed for the latest message, from the frame that started it. */ + private int open; + + private boolean terminal; + + /** Whether the end of the call was reported to the SDK; confined to {@link #serial}. */ + private boolean reported; + + private Call(Request request, Response response, String method) { + this.request = request; + this.response = response; + this.method = method; + } + + private void begin() throws IOException, InterruptedException { + // Installed first, so that a call failing below still sends its grpc-status. + response.setEntity(responseEntity()); + Metadata headers = fromHttp(request.getHeaders()); + // Runs on the connector's I/O thread as soon as the terminal write is done. + response.getDeliverySignal() + .thenAccept( + failure -> { + // An undelivered response leaves the client without the status. + if (failure != null) { + cancel( + Status.CANCELLED + .withDescription("gRPC response not delivered") + .withCause(failure)); + } else if (claim()) { + serial.execute(() -> finish(null)); + } + }); + // Bounded before the SDK sees the stream, which may run application code. + synchronized (RestletServerHost.this) { + if (!closed) { + deadline = + timer.schedule( + () -> + cancel( + Status.DEADLINE_EXCEEDED.withDescription( + "Call duration exceeded")), + duration.toNanos(), + TimeUnit.NANOSECONDS); + if (finished.get()) { + deadline.cancel(false); + } + } + } + request.getDisconnectSignal() + .thenAccept(cause -> cancel(Status.CANCELLED.withCause(cause))); + serial.execute( + () -> { + if (reported) { + return; + } + Attributes attributes = + transportListener.transportReady(transportAttributes()).toBuilder() + .set(CALL, this) + .build(); + stream = + new SdkStream( + this, + attributes, + request.getHostRef() == null + ? request.getResourceRef().getAuthority() + : request.getHostRef().getAuthority(), + StatsTraceContext.newServerContext( + tracers, method, headers)); + transportListener.streamCreated(stream, method, headers); + stream.state.onStreamAllocated(); + }); + byte[] bytes; + try (var input = request.getEntity().getStream()) { + bytes = input == null ? new byte[0] : read(input); + } + if (bytes.length > maxBytes + 5) { + cancel(Status.RESOURCE_EXHAUSTED.withDescription("Request byte budget exceeded")); + } else { + serial.execute( + () -> { + if (!reported) { + stream.state.inboundDataReceived(ReadableBuffers.wrap(bytes), true); + } + }); + } + if (!head.await(duration.toNanos(), TimeUnit.NANOSECONDS)) { + cancel(Status.DEADLINE_EXCEEDED.withDescription("Response headers timed out")); + } + } + + /** Returns the standard transport attributes: the peer and local socket addresses. */ + private Attributes transportAttributes() { + Attributes.Builder result = Attributes.newBuilder(); + SocketAddress remote = + address( + request.getClientInfo().getAddress(), + request.getClientInfo().getPort()); + if (remote != null) { + result.set(Grpc.TRANSPORT_ATTR_REMOTE_ADDR, remote); + } + SocketAddress local = + address( + response.getServerInfo().getAddress(), + response.getServerInfo().getPort()); + if (local != null) { + result.set(Grpc.TRANSPORT_ATTR_LOCAL_ADDR, local); + } + return result.build(); + } + + private byte[] read(InputStream input) throws IOException { + enterIo(); + try { + return input.readNBytes(maxBytes + 6); + } finally { + blocked.set(0); + } + } + + private OutputRepresentation responseEntity() { + return new OutputRepresentation(MediaType.valueOf("application/grpc")) { + @Override + public void write(OutputStream output) throws IOException { + try { + for (; ; ) { + byte[] data; + synchronized (queue) { + while (queue.isEmpty() && !terminal) { + queue.wait(); + } + data = queue.pollFirst(); + if (data == null) { + toHttp(trailers, response.getTrailers()); + return; + } + queued -= data.length; + } + enterIo(); + try { + output.write(data); + output.flush(); + } finally { + blocked.set(0); + } + int sent = data.length; + serial.execute(() -> stream.state.onSentBytes(sent)); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + cancel(Status.CANCELLED.withCause(e)); + throw new IOException("Interrupted writing gRPC response", e); + } catch (IOException e) { + cancel(Status.CANCELLED.withCause(e)); + throw e; + } + } + }; + } + + /** Marks the start of a network read or write, which only the peer may otherwise end. */ + private void enterIo() { + blocked.set(operations.incrementAndGet()); + if (finished.get()) { + abortIfStuck(); + } + } + + /** + * Resets the HTTP stream if the call stays in the same network read or write after it was + * ended, so a peer that stops sending or reading can't hold the handler. A call blocked + * elsewhere was already woken by the end of the call. + */ + private void abortIfStuck() { + long operation = blocked.get(); + if (operation == 0) { + return; + } + Thread.startVirtualThread( + () -> { + try { + Thread.sleep(STUCK_IO); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return; + } + if (blocked.get() == operation) { + request.abort(); + } + }); + } + + private void end(Metadata metadata) { + synchronized (queue) { + if (!terminal) { + terminal = true; + trailers = metadata; + head.countDown(); + queue.notifyAll(); + } + } + } + + private void cancel(Status status) { + if (!claim()) { + return; + } + abortIfStuck(); + // The response ends at once; only the SDK notification waits for earlier callbacks. + synchronized (queue) { + queue.clear(); + queued = 0; + open = 0; + terminal = true; + trailers = new Metadata(); + trailers.put(InternalStatus.CODE_KEY, status); + if (status.getDescription() != null) { + trailers.put(InternalStatus.MESSAGE_KEY, status.getDescription()); + } + head.countDown(); + queue.notifyAll(); + } + serial.execute(() -> finish(status)); + } + + /** + * Claims the end of the call and frees its slot at once, in the thread that ends it. + * Another thread may be draining {@link #serial}, so a slot released by {@link #finish} + * could still be held when the client sends its next call. + * + * @return True if this is the first claim. + */ + private boolean claim() { + if (!finished.compareAndSet(false, true)) { + return false; + } + calls.remove(this); + admission.release(); + // Callbacks not started by now never will, so their slot is free at once. + if (callbackState.compareAndSet(0, 2)) { + callbacks.release(); + } + ScheduledFuture timeout = deadline; + if (timeout != null) { + timeout.cancel(false); + } + return true; + } + + /** Marks the start of the application callbacks; false if the call already ended. */ + private boolean callbacksStarted() { + return callbackState.compareAndSet(0, 1); + } + + /** Releases the callback slot once, after the application callbacks exited. */ + private void callbacksEnded() { + if (callbackState.getAndSet(2) != 2) { + callbacks.release(); + } + } + + /** Reports the end of a claimed call to the SDK, in {@link #serial}. */ + private void finish(Status failure) { + reported = true; + if (deadline != null) { + deadline.cancel(false); + } + try { + if (stream != null) { + if (failure == null) { + stream.state.complete(); + } else { + stream.state.transportReportStatus(failure); + } + } + } finally { + transportListener.transportTerminated(); + } + } + + @Override + public void shutdown() {} + + @Override + public void shutdownNow(Status reason) { + cancel(reason); + } + + @Override + public ScheduledExecutorService getScheduledExecutorService() { + return sdkTimer; + } + + @Override + public ListenableFuture getStats() { + return Futures.immediateFuture(null); + } + + @Override + public InternalLogId getLogId() { + return id; + } + } + + private final class SdkStream extends AbstractServerStream { + private final Call call; + private final Attributes attributes; + private final String authority; + private final State state; + + private SdkStream( + Call call, Attributes attributes, String authority, StatsTraceContext stats) { + super(hint -> new Buffer(Math.max(1, Math.min(hint, 16384))), stats); + this.call = call; + this.attributes = attributes; + this.authority = authority; + state = new State(stats); + framer().setMaxOutboundMessageSize(maxBytes); + setOnReadyThreshold(Math.max(1, queueBytes / 2)); + } + + @Override + protected State transportState() { + return state; + } + + @Override + public Attributes getAttributes() { + return attributes; + } + + @Override + public String getAuthority() { + return authority; + } + + @Override + public int streamId() { + return -1; + } + + @Override + protected Sink abstractServerStreamSink() { + return new Sink() { + @Override + public void writeHeaders(Metadata headers, boolean flush) { + synchronized (call.queue) { + if (!call.terminal) { + toResponse(headers, call.response); + call.head.countDown(); + } + } + } + + @Override + public void writeFrame(WritableBuffer frame, boolean flush, int messages) { + if (frame == null) { + return; + } + var buffer = (Buffer) frame; + boolean overflow; + synchronized (call.queue) { + if (call.terminal) { + return; + } + // One message is framed whole, possibly before the writer starts: while the + // queue holds only that message, it may grow up to one framed message. A + // frame counting messages starts a new one. + int open = messages > 0 ? buffer.size : call.open + buffer.size; + boolean others = call.queued + buffer.size > open; + overflow = + buffer.size > queueBytes - call.queued + && (others || open > maxBytes + 5); + if (!overflow) { + call.open = open; + byte[] bytes = java.util.Arrays.copyOf(buffer.bytes, buffer.size); + onSendingBytes(bytes.length); + call.queue.addLast(bytes); + call.queued += bytes.length; + call.queue.notifyAll(); + } + } + if (overflow) { + call.cancel( + Status.RESOURCE_EXHAUSTED.withDescription( + "Response queue byte budget exceeded")); + } + } + + @Override + public void writeTrailers(Metadata trailers, boolean headersSent, Status status) { + call.end(trailers); + } + + @Override + public void cancel(Status status) { + call.cancel(status); + } + }; + } + + private final class State extends AbstractServerStream.TransportState { + private State(StatsTraceContext stats) { + super(maxBytes, stats, new TransportTracer()); + } + + @Override + public void runOnTransportThread(Runnable task) { + call.serial.execute(task); + } + + @Override + public void bytesRead(int count) {} + + @Override + public void deframeFailed(Throwable cause) { + call.cancel(Status.fromThrowable(cause)); + } + } + } + + private static final class Buffer implements WritableBuffer { + private final byte[] bytes; + private int size; + + private Buffer(int capacity) { + bytes = new byte[capacity]; + } + + @Override + public void write(byte value) { + bytes[size++] = value; + } + + @Override + public void write(byte[] source, int offset, int count) { + System.arraycopy(source, offset, bytes, size, count); + size += count; + } + + @Override + public int writableBytes() { + return bytes.length - size; + } + + @Override + public int readableBytes() { + return size; + } + + @Override + public void release() {} + } + + /** Returns the socket address of an IP literal and port, or null if unknown. */ + private static SocketAddress address(String host, int port) { + if (host == null || port < 0) { + return null; + } + try { + return new InetSocketAddress(InetAddress.ofLiteral(host), port); + } catch (IllegalArgumentException e) { + return null; + } + } + + private static Metadata fromHttp(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))); + } + + /** + * Maps initial metadata onto the response as raw headers, written as-is whether or not their + * names are standard HTTP ones, with their repeated values. + */ + private static void toResponse(Metadata metadata, Response response) { + var raw = new Series
(Header.class); + toHttp(metadata, raw); + raw.removeIf(header -> FRAMING_HEADERS.contains(header.getName())); + if (!raw.isEmpty()) { + response.getAttributes().put(HeaderConstants.ATTRIBUTE_RAW_RESPONSE_HEADERS, raw); + } + } + + private static void toHttp(Metadata metadata, Series
headers) { + if (metadata == null) { + return; + } + byte[][] values = TransportFrameUtil.toHttp2Headers(metadata); + for (int i = 0; i < values.length; i += 2) { + String name = new String(values[i], StandardCharsets.US_ASCII); + if (!name.startsWith(":") && !"content-type".equals(name)) { + headers.add(name, new String(values[i + 1], StandardCharsets.US_ASCII)); + } + } + } +} 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 bf06dc2919..5b19cde07b 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 @@ -281,22 +281,32 @@ public ServerCall.Listener interceptCall( 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()); + boolean rejected = false; + try { + call.start( + new ClientCall.Listener<>() { + @Override + public void onClose(Status value, Metadata trailers) { + status.set(value); + terminals.incrementAndGet(); + closed.countDown(); + } + }, + new Metadata()); + } catch (IllegalStateException e) { + // A call created once the channel has resolved its target rejects start() after + // cancel(); one created earlier is delayed and closes its listener instead. + assertEquals("call was cancelled", e.getMessage()); + rejected = true; + } + if (!rejected) { + 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()); + assertEquals(rejected ? 0 : 1, terminals.get()); } @Test 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 index 81ff8796d1..e9a6c76a42 100644 --- 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 @@ -24,7 +24,7 @@ class GrpcModuleTestCase { @Test - void nativeChannelResolvesWithoutSdkNetworkTransportOrInternalExports(@TempDir Path temp) + void nativeAdaptersResolveWithoutSdkNetworkTransportOrInternalExports(@TempDir Path temp) throws Exception { String classPath = System.getProperty( @@ -79,12 +79,61 @@ void nativeChannelResolvesWithoutSdkNetworkTransportOrInternalExports(@TempDir P import org.restlet.Context; import org.restlet.data.Protocol; import org.restlet.ext.grpc.GrpcClientAdapter; + import org.restlet.ext.grpc.GrpcServerAdapter; + import org.restlet.routing.Router; + import io.grpc.CallOptions; + import io.grpc.ClientCall; + import io.grpc.Metadata; + import io.grpc.MethodDescriptor; + import io.grpc.ServerCall; + import io.grpc.ServerServiceDefinition; + import io.grpc.Status; 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 marshaller = new MethodDescriptor.Marshaller() { + public java.io.InputStream stream(String value) { + return new java.io.ByteArrayInputStream( + value.getBytes(java.nio.charset.StandardCharsets.UTF_8)); + } + public String parse(java.io.InputStream input) { + try { + return new String(input.readAllBytes(), + java.nio.charset.StandardCharsets.UTF_8); + } catch (java.io.IOException e) { + throw new java.io.UncheckedIOException(e); + } + } + }; + var method = MethodDescriptor.newBuilder() + .setType(MethodDescriptor.MethodType.UNARY) + .setFullMethodName("consumer.Echo/Call") + .setRequestMarshaller(marshaller) + .setResponseMarshaller(marshaller).build(); + var server = new GrpcServerAdapter(); + server.getServices().add(ServerServiceDefinition.builder("consumer.Echo") + .addMethod(method, (call, headers) -> { + call.request(1); + return new ServerCall.Listener() { + private String message; + public void onMessage(String value) { message = value; } + public void onHalfClose() { + call.sendHeaders(new Metadata()); + call.sendMessage(message); + call.close(Status.OK, new Metadata()); + } + }; + }).build()); + server.start(); + var router = new Router(); + server.attachTo(router); + var serverContext = new Context(); + serverContext.getParameters().set("http.transport.protocol", "HTTP2"); + var http = new org.restlet.Server(serverContext, Protocol.HTTP, 0, router); + http.start(); var context = new Context(); context.getParameters().set("httpClientTransportMode", "HTTP2"); context.getParameters().set("followRedirects", "false"); @@ -93,12 +142,44 @@ public static void main(String[] args) throws Exception { try (var adapter = new GrpcClientAdapter()) { adapter.setClient(client); adapter.setPlaintext(true); - adapter.setTarget("dns:///localhost:12345"); + String authority = "localhost:" + http.getActualPort(); + adapter.setTarget("dns:///" + authority); adapter.start(); - if (!adapter.getChannel().authority().equals("localhost:12345")) { + if (!adapter.getChannel().authority().equals(authority)) { throw new AssertionError("Native authority was lost"); } - } finally { client.stop(); } + var done = new java.util.concurrent.CountDownLatch(1); + var reply = new java.util.concurrent.atomic.AtomicReference(); + var result = new java.util.concurrent.atomic.AtomicReference(); + var call = adapter.getChannel().newCall(method, + CallOptions.DEFAULT.withDeadlineAfter(10, + java.util.concurrent.TimeUnit.SECONDS)); + call.start(new ClientCall.Listener() { + public void onMessage(String value) { reply.set(value); } + public void onClose(Status status, Metadata trailers) { + result.set(status); + done.countDown(); + } + }, new Metadata()); + call.request(1); + call.sendMessage("packaged echo"); + call.halfClose(); + if (!done.await(15, java.util.concurrent.TimeUnit.SECONDS) + || result.get() == null || !result.get().isOk() + || !"packaged echo".equals(reply.get())) { + throw new AssertionError("Native round trip failed: " + result.get()); + } + server.stop(); + if (!http.isStarted()) { + throw new AssertionError("Borrowed listener was stopped"); + } + } finally { + try { server.stop(); } + finally { + try { client.stop(); } + finally { http.stop(); } + } + } } } """); diff --git a/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcServerAdapterTestCase.java b/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcServerAdapterTestCase.java new file mode 100644 index 0000000000..434bfa6dd6 --- /dev/null +++ b/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcServerAdapterTestCase.java @@ -0,0 +1,1348 @@ +/** + * 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.assertFalse; +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.ManagedChannel; +import io.grpc.Metadata; +import io.grpc.MethodDescriptor; +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.NettyChannelBuilder; +import io.grpc.protobuf.ProtoUtils; +import io.grpc.protobuf.services.HealthStatusManager; +import io.grpc.stub.ClientCalls; +import io.grpc.stub.ServerCallStreamObserver; +import io.grpc.stub.ServerCalls; +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.DataOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import java.util.logging.Handler; +import java.util.logging.Level; +import java.util.logging.LogRecord; +import java.util.logging.Logger; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +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.InputRepresentation; +import org.restlet.routing.Router; + +class GrpcServerAdapterTestCase { + private static final MethodDescriptor ECHO = + MethodDescriptor.newBuilder() + .setType(MethodDescriptor.MethodType.UNARY) + .setFullMethodName("review.Server/Echo") + .setRequestMarshaller(ProtoUtils.marshaller(StringValue.getDefaultInstance())) + .setResponseMarshaller(ProtoUtils.marshaller(StringValue.getDefaultInstance())) + .build(); + private static final MethodDescriptor WATCH = + ECHO.toBuilder() + .setType(MethodDescriptor.MethodType.SERVER_STREAMING) + .setFullMethodName("review.Server/Watch") + .build(); + private final GrpcServerAdapter adapter = new GrpcServerAdapter(); + private Server http; + private ManagedChannel channel; + private final java.util.List lifecycleErrors = new CopyOnWriteArrayList<>(); + private final Handler errorHandler = + new Handler() { + @Override + public void publish(LogRecord record) { + if (record.getLevel().intValue() >= Level.SEVERE.intValue()) { + lifecycleErrors.add(record); + } + } + + @Override + public void flush() {} + + @Override + public void close() {} + }; + + @BeforeEach + void captureLifecycleErrors() { + Logger.getLogger("io.grpc.internal").addHandler(errorHandler); + } + + private void start(ServerServiceDefinition service) throws Exception { + adapter.getServices().add(service); + adapter.start(); + Router router = new Router(); + adapter.attachTo(router); + router.attach( + "/rest", + new Restlet() { + @Override + public void handle(Request request, Response response) { + response.setEntity("rest", MediaType.TEXT_PLAIN); + } + }); + Context context = new Context(); + context.getParameters().set("http.transport.protocol", "HTTP2"); + http = new Server(context, Protocol.HTTP, 0, router); + http.start(); + channel = + NettyChannelBuilder.forAddress("localhost", http.getActualPort()) + .usePlaintext() + .build(); + } + + private String call(String value) { + return ClientCalls.blockingUnaryCall( + channel, + ECHO, + CallOptions.DEFAULT.withDeadlineAfter(10, TimeUnit.SECONDS), + StringValue.of(value)) + .getValue(); + } + + private static ServerServiceDefinition echo() { + return ServerServiceDefinition.builder("review.Server") + .addMethod( + ECHO, + ServerCalls.asyncUnaryCall( + (StringValue request, + io.grpc.stub.StreamObserver out) -> { + out.onNext(request); + out.onCompleted(); + })) + .build(); + } + + @AfterEach + void cleanup() throws Exception { + try { + if (channel != null) { + channel.shutdownNow(); + assertTrue(channel.awaitTermination(15, TimeUnit.SECONDS)); + } + } finally { + try { + adapter.stop(); + } finally { + if (http != null) { + http.stop(); + } + Logger.getLogger("io.grpc.internal").removeHandler(errorHandler); + assertTrue( + lifecycleErrors.isEmpty(), + () -> + lifecycleErrors.stream() + .map(LogRecord::getMessage) + .toList() + .toString()); + } + } + } + + @Test + void generatedAndDynamicUnaryShareListenerWithRest() throws Exception { + adapter.getServices().add(new HealthStatusManager().getHealthService().bindService()); + start(echo()); + assertEquals("hello", call("hello")); + assertEquals( + HealthCheckResponse.ServingStatus.SERVING, + HealthGrpc.newBlockingStub(channel) + .withDeadlineAfter(10, TimeUnit.SECONDS) + .check(HealthCheckRequest.getDefaultInstance()) + .getStatus()); + var method = + MethodDescriptor.newBuilder() + .setType(MethodDescriptor.MethodType.UNARY) + .setFullMethodName(ECHO.getFullMethodName()) + .setRequestMarshaller( + ProtoUtils.marshaller( + DynamicMessage.getDefaultInstance( + StringValue.getDescriptor()))) + .setResponseMarshaller( + ProtoUtils.marshaller( + DynamicMessage.getDefaultInstance( + StringValue.getDescriptor()))) + .build(); + var request = + DynamicMessage.newBuilder(StringValue.getDescriptor()) + .setField(StringValue.getDescriptor().findFieldByName("value"), "dynamic") + .build(); + assertEquals( + request, + ClientCalls.blockingUnaryCall( + channel, + method, + CallOptions.DEFAULT.withDeadlineAfter(10, TimeUnit.SECONDS), + request)); + Context context = new Context(); + context.getParameters().set("httpClientTransportMode", "HTTP2"); + Client rest = new Client(context, Protocol.HTTP); + rest.start(); + try { + assertEquals( + "rest", + rest.handle( + new Request( + org.restlet.data.Method.GET, + "http://localhost:" + http.getActualPort() + "/rest")) + .getEntity() + .getText()); + adapter.stop(); + assertTrue(http.isStarted()); + assertEquals( + "rest", + rest.handle( + new Request( + org.restlet.data.Method.GET, + "http://localhost:" + http.getActualPort() + "/rest")) + .getEntity() + .getText()); + } finally { + rest.stop(); + } + } + + @Test + void preservesTrailersOnlyErrorsAndRecovers() throws Exception { + Metadata.Key key = Metadata.Key.of("x-detail", Metadata.ASCII_STRING_MARSHALLER); + start( + ServerServiceDefinition.builder("review.Server") + .addMethod( + ECHO, + ServerCalls.asyncUnaryCall( + (StringValue request, + io.grpc.stub.StreamObserver out) -> { + Metadata trailers = new Metadata(); + trailers.put(key, "one"); + trailers.put(key, "two"); + out.onError( + Status.INVALID_ARGUMENT + .withDescription("native % error") + .asRuntimeException(trailers)); + })) + .build()); + var error = assertThrows(StatusRuntimeException.class, () -> call("bad")); + assertEquals(Status.Code.INVALID_ARGUMENT, error.getStatus().getCode()); + assertEquals("native % error", error.getStatus().getDescription()); + assertEquals( + java.util.List.of("one", "two"), + com.google.common.collect.ImmutableList.copyOf( + java.util.Objects.requireNonNull(error.getTrailers()).getAll(key))); + } + + @Test + void serverStreamingUsesReadinessAndPreservesOrder() throws Exception { + adapter.setMaxQueuedBytes(32768); + AtomicInteger produced = new AtomicInteger(); + start( + ServerServiceDefinition.builder("review.Server") + .addMethod( + WATCH, + ServerCalls.asyncServerStreamingCall( + (StringValue request, + io.grpc.stub.StreamObserver out) -> { + var observer = + (ServerCallStreamObserver) out; + observer.setOnReadyHandler( + () -> { + while (observer.isReady() + && produced.get() < 100) { + int index = produced.getAndIncrement(); + observer.onNext( + StringValue.of( + index + + ":" + + "x" + .repeat( + 4096))); + } + if (produced.compareAndSet(100, 101)) { + observer.onCompleted(); + } + }); + })) + .build()); + var responses = + ClientCalls.blockingServerStreamingCall( + channel, + WATCH, + CallOptions.DEFAULT.withDeadlineAfter(15, TimeUnit.SECONDS), + StringValue.of("watch")); + var values = new ArrayList(); + while (responses.hasNext()) { + values.add(responses.next().getValue()); + Thread.sleep(2); + } + assertEquals(100, values.size()); + for (int i = 0; i < 100; i++) { + assertEquals(i + ":" + "x".repeat(4096), values.get(i)); + } + } + + @Test + void cancellationAndShutdownReleaseAdmissionWithoutStoppingListener() throws Exception { + adapter.setMaxConcurrentCalls(1); + adapter.setShutdownTimeout(Duration.ofMillis(100)); + CountDownLatch entered = new CountDownLatch(1); + CountDownLatch cancelled = new CountDownLatch(1); + start( + ServerServiceDefinition.builder("review.Server") + .addMethod( + ECHO, + ServerCalls.asyncUnaryCall( + (StringValue request, + io.grpc.stub.StreamObserver out) -> { + if ("block".equals(request.getValue())) { + ((ServerCallStreamObserver) out) + .setOnCancelHandler(cancelled::countDown); + entered.countDown(); + } else { + out.onNext(request); + out.onCompleted(); + } + })) + .build()); + var first = + ClientCalls.futureUnaryCall( + channel.newCall(ECHO, CallOptions.DEFAULT), StringValue.of("block")); + assertTrue(entered.await(10, TimeUnit.SECONDS)); + first.cancel(true); + assertTrue(cancelled.await(10, TimeUnit.SECONDS)); + assertEquals("next", call("next")); + var second = + ClientCalls.futureUnaryCall( + channel.newCall(ECHO, CallOptions.DEFAULT), StringValue.of("block")); + adapter.stop(); + assertThrows( + java.util.concurrent.ExecutionException.class, + () -> second.get(10, TimeUnit.SECONDS)); + assertTrue(http.isStarted()); + assertFalse(adapter.isStarted()); + } + + @Test + void requestAndResponseMessageBudgetsAreEnforced() throws Exception { + adapter.setMaxMessageBytes(128); + start(echo()); + assertEquals("x".repeat(126), call("x".repeat(126))); + assertEquals( + Status.Code.RESOURCE_EXHAUSTED, + assertThrows(StatusRuntimeException.class, () -> call("x".repeat(127))) + .getStatus() + .getCode()); + assertEquals("next", call("next")); + } + + @Test + void unsupportedStreamingRegistrationFailsExplicitly() { + adapter.getServices() + .add( + ServerServiceDefinition.builder("review.Server") + .addMethod( + ECHO.toBuilder() + .setType( + MethodDescriptor.MethodType + .CLIENT_STREAMING) + .build(), + ServerCalls.asyncClientStreamingCall( + out -> + new io.grpc.stub.StreamObserver< + StringValue>() { + @Override + public void onNext(StringValue value) {} + + @Override + public void onError(Throwable cause) {} + + @Override + public void onCompleted() {} + })) + .build()); + assertThrows(UnsupportedOperationException.class, adapter::start); + } + + private static byte[] frame(String value) throws IOException { + var bytes = new ByteArrayOutputStream(); + var data = new DataOutputStream(bytes); + byte[] message = StringValue.of(value).toByteArray(); + data.writeByte(0); + data.writeInt(message.length); + data.write(message); + return bytes.toByteArray(); + } + + /** A direct HTTP/2 gRPC request; its abort action, if any, stands for the stream reset. */ + private static Request grpcRequest( + MethodDescriptor method, InputStream body, Runnable abort) { + Request request = + new Request( + org.restlet.data.Method.POST, + "http://localhost/" + method.getFullMethodName()) { + @Override + public boolean abort() { + abort.run(); + return true; + } + }; + request.setProtocol(Protocol.valueOf("HTTP", "2.0")); + request.setEntity(new InputRepresentation(body, MediaType.valueOf("application/grpc"))); + return request; + } + + private Response dispatch(MethodDescriptor method, String value) + throws Exception { + Request request = grpcRequest(method, new ByteArrayInputStream(frame(value)), () -> {}); + Response response = new Response(request); + adapter.handle(request, response); + return response; + } + + private static String trailersAfterWriting(Response response) throws IOException { + response.getEntity().write(new ByteArrayOutputStream()); + return response.getTrailers().getFirstValue("grpc-status"); + } + + private static ServerServiceDefinition counting(AtomicInteger calls) { + return ServerServiceDefinition.builder("review.Server") + .addMethod( + ECHO, + ServerCalls.asyncUnaryCall( + (StringValue request, + io.grpc.stub.StreamObserver out) -> { + calls.incrementAndGet(); + out.onNext(request); + out.onCompleted(); + })) + .build(); + } + + @Test + void dotSegmentsCannotReachAMethodOutsideTheirRoute() throws Exception { + AtomicInteger calls = new AtomicInteger(); + adapter.getServices().add(counting(calls)); + adapter.start(); + Request request = grpcRequest(ECHO, new ByteArrayInputStream(frame("x")), () -> {}); + request.setResourceRef("http://localhost/public.Service/../review.Server/Echo"); + Response response = new Response(request); + adapter.handle(request, response); + assertEquals(org.restlet.data.Status.CLIENT_ERROR_NOT_FOUND, response.getStatus()); + assertEquals(0, calls.get()); + } + + @Test + void grpcMediaTypeParametersAreAccepted() throws Exception { + AtomicInteger calls = new AtomicInteger(); + adapter.getServices().add(counting(calls)); + adapter.start(); + Request request = grpcRequest(ECHO, new ByteArrayInputStream(frame("x")), () -> {}); + request.getEntity().setMediaType(MediaType.valueOf("application/grpc; version=1")); + Response response = new Response(request); + adapter.handle(request, response); + assertEquals("0", trailersAfterWriting(response)); + assertEquals(1, calls.get()); + } + + private void assertBlockedDirectCallbackEndsAtTheDeadline(boolean inInterceptor) + throws Exception { + CountDownLatch release = new CountDownLatch(1); + CountDownLatch entered = new CountDownLatch(1); + Runnable block = + () -> { + entered.countDown(); + try { + release.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + }; + adapter.setExecutor(Runnable::run); + adapter.setMaxCallDuration(Duration.ofMillis(300)); + adapter.setMaxConcurrentCalls(1); + if (inInterceptor) { + adapter.getServices().add(echo()); + adapter.getInterceptors() + .add( + new io.grpc.ServerInterceptor() { + @Override + public io.grpc.ServerCall.Listener interceptCall( + io.grpc.ServerCall call, + Metadata headers, + io.grpc.ServerCallHandler next) { + block.run(); + return next.startCall(call, headers); + } + }); + } else { + adapter.getServices() + .add( + ServerServiceDefinition.builder("review.Server") + .addMethod( + ECHO, + ServerCalls.asyncUnaryCall( + (StringValue request, + io.grpc.stub.StreamObserver + out) -> { + block.run(); + out.onNext(request); + out.onCompleted(); + })) + .build()); + } + adapter.start(); + try (var executor = Executors.newVirtualThreadPerTaskExecutor()) { + // The call limit is 1, so blocked callbacks may hold up to 2 callback slots. + for (int i = 0; i < 2; i++) { + Response response = + executor.submit(() -> dispatch(ECHO, "blocked")).get(10, TimeUnit.SECONDS); + assertTrue(entered.await(10, TimeUnit.SECONDS)); + assertEquals( + String.valueOf(Status.Code.DEADLINE_EXCEEDED.value()), + trailersAfterWriting(response)); + } + assertEquals( + org.restlet.data.Status.SERVER_ERROR_SERVICE_UNAVAILABLE, + dispatch(ECHO, "bounded").getStatus()); + release.countDown(); + long limit = System.nanoTime() + TimeUnit.SECONDS.toNanos(10); + Response admitted = dispatch(ECHO, "admitted"); + while (org.restlet.data.Status.SERVER_ERROR_SERVICE_UNAVAILABLE.equals( + admitted.getStatus()) + && System.nanoTime() < limit) { + Thread.onSpinWait(); + admitted = dispatch(ECHO, "admitted"); + } + assertEquals("0", trailersAfterWriting(admitted)); + } finally { + release.countDown(); + } + } + + @Test + void blockingInterceptorWithADirectExecutorEndsAtTheDeadline() throws Exception { + assertBlockedDirectCallbackEndsAtTheDeadline(true); + } + + @Test + void blockingHandlerWithADirectExecutorEndsAtTheDeadline() throws Exception { + assertBlockedDirectCallbackEndsAtTheDeadline(false); + } + + @Test + void stopReportsTheTimeoutWhenACancellationListenerBlocks() throws Exception { + adapter.setShutdownTimeout(Duration.ofMillis(300)); + CountDownLatch entered = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + start( + ServerServiceDefinition.builder("review.Server") + .addMethod( + ECHO, + ServerCalls.asyncUnaryCall( + (StringValue request, + io.grpc.stub.StreamObserver out) -> { + io.grpc.Context.current() + .addListener( + context -> { + try { + release.await(); + } catch (InterruptedException e) { + Thread.currentThread() + .interrupt(); + } + }, + Runnable::run); + entered.countDown(); + })) + .build()); + ClientCalls.futureUnaryCall( + channel.newCall(ECHO, CallOptions.DEFAULT), StringValue.of("block")); + assertTrue(entered.await(10, TimeUnit.SECONDS)); + try { + long started = System.nanoTime(); + assertTrue( + assertThrows(IllegalStateException.class, adapter::stop) + .getMessage() + .endsWith("shutdown timed out")); + assertTrue(System.nanoTime() - started < TimeUnit.SECONDS.toNanos(5)); + assertFalse(adapter.isStarted()); + } finally { + release.countDown(); + } + } + + @Test + void consecutiveCallsAtTheLimitAreAdmittedOnceDelivered() throws Exception { + adapter.setMaxConcurrentCalls(1); + start(echo()); + for (int i = 0; i < 50; i++) { + assertEquals("call-" + i, call("call-" + i)); + } + } + + @Test + void initialMetadataWithStandardHeaderNamesIsSent() throws Exception { + Metadata.Key cacheControl = + Metadata.Key.of("cache-control", Metadata.ASCII_STRING_MARSHALLER); + Metadata.Key etag = Metadata.Key.of("etag", Metadata.ASCII_STRING_MARSHALLER); + Metadata.Key custom = Metadata.Key.of("x-detail", Metadata.ASCII_STRING_MARSHALLER); + Metadata.Key allow = Metadata.Key.of("allow", Metadata.ASCII_STRING_MARSHALLER); + Metadata.Key expose = + Metadata.Key.of("access-control-expose-headers", Metadata.ASCII_STRING_MARSHALLER); + adapter.getInterceptors() + .add( + new io.grpc.ServerInterceptor() { + @Override + public io.grpc.ServerCall.Listener interceptCall( + io.grpc.ServerCall call, + Metadata headers, + io.grpc.ServerCallHandler next) { + return next.startCall( + new io.grpc.ForwardingServerCall + .SimpleForwardingServerCall<>(call) { + @Override + public void sendHeaders(Metadata metadata) { + metadata.put(cacheControl, "no-cache"); + metadata.put(cacheControl, "no-store"); + metadata.put(etag, "\"v1\""); + metadata.put(custom, "kept"); + metadata.put(allow, "POST"); + metadata.put(expose, "x-detail"); + super.sendHeaders(metadata); + } + }, + headers); + } + }); + start(echo()); + AtomicReference received = new AtomicReference<>(); + AtomicReference trailers = new AtomicReference<>(); + var intercepted = + io.grpc.ClientInterceptors.intercept( + channel, + io.grpc.stub.MetadataUtils.newCaptureMetadataInterceptor( + received, trailers)); + assertEquals( + "meta", + ClientCalls.blockingUnaryCall( + intercepted, + ECHO, + CallOptions.DEFAULT.withDeadlineAfter(10, TimeUnit.SECONDS), + StringValue.of("meta")) + .getValue()); + String directives = String.join(",", received.get().getAll(cacheControl)); + assertTrue(directives.contains("no-cache"), directives); + assertTrue(directives.contains("no-store"), directives); + assertEquals("\"v1\"", received.get().get(etag)); + assertEquals("kept", received.get().get(custom)); + assertEquals("POST", received.get().get(allow)); + assertEquals("x-detail", received.get().get(expose)); + var values = new ArrayList(); + received.get().getAll(cacheControl).forEach(values::add); + assertEquals(List.of("no-cache", "no-store"), values); + } + + @Test + void streamingErrorAfterMessagesKeepsItsStatusAndTrailers() throws Exception { + Metadata.Key key = Metadata.Key.of("x-detail", Metadata.ASCII_STRING_MARSHALLER); + start( + ServerServiceDefinition.builder("review.Server") + .addMethod( + WATCH, + ServerCalls.asyncServerStreamingCall( + (StringValue request, + io.grpc.stub.StreamObserver out) -> { + out.onNext(StringValue.of("first")); + out.onNext(StringValue.of("second")); + Metadata trailers = new Metadata(); + trailers.put(key, "late"); + out.onError( + Status.ABORTED + .withDescription("after messages") + .asRuntimeException(trailers)); + })) + .build()); + var responses = + ClientCalls.blockingServerStreamingCall( + channel, + WATCH, + CallOptions.DEFAULT.withDeadlineAfter(10, TimeUnit.SECONDS), + StringValue.of("watch")); + assertEquals("first", responses.next().getValue()); + assertEquals("second", responses.next().getValue()); + var error = assertThrows(StatusRuntimeException.class, responses::hasNext); + assertEquals(Status.Code.ABORTED, error.getStatus().getCode()); + assertEquals("after messages", error.getStatus().getDescription()); + assertEquals("late", java.util.Objects.requireNonNull(error.getTrailers()).get(key)); + } + + @Test + void transportAttributesCarryTheSocketAddresses() throws Exception { + AtomicReference seen = new AtomicReference<>(); + adapter.getInterceptors() + .add( + new io.grpc.ServerInterceptor() { + @Override + public io.grpc.ServerCall.Listener interceptCall( + io.grpc.ServerCall call, + Metadata headers, + io.grpc.ServerCallHandler next) { + seen.set(call.getAttributes()); + return next.startCall(call, headers); + } + }); + start(echo()); + assertEquals("peer", call("peer")); + var remote = + (java.net.InetSocketAddress) + seen.get().get(io.grpc.Grpc.TRANSPORT_ATTR_REMOTE_ADDR); + var local = + (java.net.InetSocketAddress) seen.get().get(io.grpc.Grpc.TRANSPORT_ATTR_LOCAL_ADDR); + assertTrue(remote.getAddress().isLoopbackAddress(), remote::toString); + assertTrue(remote.getPort() > 0); + assertTrue(local.getAddress().isLoopbackAddress(), local::toString); + assertEquals(http.getActualPort(), local.getPort()); + } + + @Test + void blockingClientDeadlineListenerCannotDelayCallBudgets() throws Exception { + CountDownLatch entered = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + adapter.setMaxCallDuration(Duration.ofSeconds(1)); + adapter.getServices() + .add( + ServerServiceDefinition.builder("review.Server") + .addMethod( + ECHO, + (call, headers) -> { + // Runs where the client deadline expires. + io.grpc.Context.current() + .addListener( + context -> { + entered.countDown(); + try { + release.await(); + } catch (InterruptedException e) { + Thread.currentThread() + .interrupt(); + } + }, + Runnable::run); + call.sendHeaders(new Metadata()); + return new io.grpc.ServerCall.Listener<>() {}; + }) + .addMethod( + WATCH, + (call, headers) -> { + call.sendHeaders(new Metadata()); + return new io.grpc.ServerCall.Listener<>() {}; + }) + .build()); + adapter.start(); + // Released before the executor waits for its tasks, which the listener may hold. + try (var executor = Executors.newVirtualThreadPerTaskExecutor()) { + try { + Request blocking = + grpcRequest(ECHO, new ByteArrayInputStream(frame("x")), () -> {}); + blocking.getHeaders().add("grpc-timeout", "100m"); + executor.submit(() -> adapter.handle(blocking, new Response(blocking))); + assertTrue(entered.await(10, TimeUnit.SECONDS)); + Response budgeted = dispatch(WATCH, "y"); + assertEquals( + String.valueOf(Status.Code.DEADLINE_EXCEEDED.value()), + executor.submit(() -> trailersAfterWriting(budgeted)) + .get(5, TimeUnit.SECONDS)); + } finally { + release.countDown(); + } + } + } + + @Test + void http1RequestsAreRejectedBeforeDispatch() throws Exception { + AtomicInteger calls = new AtomicInteger(); + adapter.getServices() + .add( + ServerServiceDefinition.builder("review.Server") + .addMethod( + ECHO, + ServerCalls.asyncUnaryCall( + (StringValue request, + io.grpc.stub.StreamObserver + out) -> { + calls.incrementAndGet(); + out.onNext(request); + out.onCompleted(); + })) + .build()); + adapter.start(); + Request request = grpcRequest(ECHO, new ByteArrayInputStream(frame("x")), () -> {}); + request.setProtocol(Protocol.valueOf("HTTP", "1.1")); + Response response = new Response(request); + adapter.handle(request, response); + assertEquals( + org.restlet.data.Status.SERVER_ERROR_VERSION_NOT_SUPPORTED, response.getStatus()); + assertEquals(0, calls.get()); + } + + @Test + void unreadableRequestStillSendsItsGrpcStatus() throws Exception { + adapter.getServices().add(echo()); + adapter.start(); + Request request = + grpcRequest( + ECHO, + new InputStream() { + @Override + public int read() throws IOException { + throw new IOException("broken body"); + } + }, + () -> {}); + Response response = new Response(request); + adapter.handle(request, response); + assertEquals( + String.valueOf(Status.Code.INVALID_ARGUMENT.value()), + trailersAfterWriting(response)); + } + + @Test + void stalledRequestBodyIsResetAtTheDeadline() throws Exception { + adapter.setMaxCallDuration(Duration.ofMillis(300)); + adapter.getServices().add(echo()); + adapter.start(); + CountDownLatch reset = new CountDownLatch(1); + InputStream stalled = + new InputStream() { + @Override + public int read() throws IOException { + try { + reset.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + throw new IOException("stream reset"); + } + }; + Request request = grpcRequest(ECHO, stalled, reset::countDown); + Response response = new Response(request); + try (var executor = Executors.newVirtualThreadPerTaskExecutor()) { + executor.submit(() -> adapter.handle(request, response)).get(10, TimeUnit.SECONDS); + } + assertEquals(0, reset.getCount()); + assertEquals( + String.valueOf(Status.Code.DEADLINE_EXCEEDED.value()), + trailersAfterWriting(response)); + } + + @Test + void responseWriteBlockedByThePeerIsResetAtShutdown() throws Exception { + adapter.getServices() + .add( + ServerServiceDefinition.builder("review.Server") + .addMethod( + WATCH, + ServerCalls.asyncServerStreamingCall( + (StringValue request, + io.grpc.stub.StreamObserver + out) -> + out.onNext(StringValue.of("first")))) + .build()); + adapter.start(); + CountDownLatch reset = new CountDownLatch(1); + Request request = + grpcRequest(WATCH, new ByteArrayInputStream(frame("watch")), reset::countDown); + Response response = new Response(request); + adapter.handle(request, response); + CountDownLatch writing = new CountDownLatch(1); + try (var executor = Executors.newVirtualThreadPerTaskExecutor()) { + var writer = + executor.submit( + () -> { + response.getEntity() + .write( + new OutputStream() { + @Override + public void write(int value) + throws IOException { + writing.countDown(); + try { + reset.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + throw new IOException("stream reset"); + } + }); + return null; + }); + assertTrue(writing.await(10, TimeUnit.SECONDS)); + adapter.stop(); + assertTrue(reset.await(10, TimeUnit.SECONDS)); + assertThrows( + java.util.concurrent.ExecutionException.class, + () -> writer.get(10, TimeUnit.SECONDS)); + } + } + + @Test + void callbacksMayUseTheAdapterWhileItStops() throws Exception { + CountDownLatch entered = new CountDownLatch(1); + start( + ServerServiceDefinition.builder("review.Server") + .addMethod( + ECHO, + ServerCalls.asyncUnaryCall( + (StringValue request, + io.grpc.stub.StreamObserver out) -> { + CountDownLatch cancelled = new CountDownLatch(1); + io.grpc.Context.current() + .addListener( + context -> cancelled.countDown(), + Runnable::run); + entered.countDown(); + try { + cancelled.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + adapter.getMaxConcurrentCalls(); + })) + .build()); + ClientCalls.futureUnaryCall( + channel.newCall(ECHO, CallOptions.DEFAULT), StringValue.of("block")); + assertTrue(entered.await(10, TimeUnit.SECONDS)); + adapter.stop(); + assertFalse(adapter.isStarted()); + assertTrue(http.isStarted()); + } + + @Test + void unaryResponseLargerThanTheQueueBudgetFitsWithADirectExecutor() throws Exception { + adapter.setExecutor(Runnable::run); + String large = "x".repeat(512 * 1024); + start( + ServerServiceDefinition.builder("review.Server") + .addMethod( + ECHO, + ServerCalls.asyncUnaryCall( + (StringValue request, + io.grpc.stub.StreamObserver out) -> { + out.onNext(StringValue.of(large)); + out.onCompleted(); + })) + .build()); + assertEquals(large, call("large")); + } + + @Test + void completedResponseRetainsAdmissionUntilSentAndBorrowedExecutorSurvives() throws Exception { + try (var executor = Executors.newVirtualThreadPerTaskExecutor()) { + adapter.setExecutor(executor); + adapter.setMaxConcurrentCalls(1); + adapter.getServices().add(echo()); + adapter.start(); + Response first = dispatch(ECHO, "first"); + first.getEntity().write(OutputStream.nullOutputStream()); + assertEquals( + org.restlet.data.Status.SERVER_ERROR_SERVICE_UNAVAILABLE, + dispatch(ECHO, "busy").getStatus()); + first.notifyDelivered(null); + first.notifyDelivered(null); + Response next = dispatch(ECHO, "next"); + next.getEntity().write(OutputStream.nullOutputStream()); + next.notifyDelivered(null); + adapter.stop(); + adapter.stop(); + assertFalse(executor.isShutdown()); + assertEquals("still available", executor.submit(() -> "still available").get()); + } + } + + @Test + void stalledWriterBoundsReadinessAndResumesInOrder() throws Exception { + int queueBudget = 32768; + adapter.setMaxQueuedBytes(queueBudget); + AtomicInteger produced = new AtomicInteger(); + CountDownLatch backpressured = new CountDownLatch(1); + AtomicReference> observerRef = + new AtomicReference<>(); + adapter.getServices() + .add( + ServerServiceDefinition.builder("review.Server") + .addMethod( + WATCH, + ServerCalls.asyncServerStreamingCall( + (StringValue request, + io.grpc.stub.StreamObserver + out) -> { + var observer = + (ServerCallStreamObserver) + out; + observerRef.set(observer); + observer.setOnReadyHandler( + () -> { + while (observer.isReady() + && produced.get() < 100) { + observer.onNext( + StringValue.of( + produced + .getAndIncrement() + + ":" + + "x" + .repeat( + 4096))); + } + if (produced.compareAndSet( + 100, 101)) { + observer.onCompleted(); + } else if (!observer.isReady()) { + backpressured.countDown(); + } + }); + })) + .build()); + adapter.start(); + Response response = dispatch(WATCH, "watch"); + CountDownLatch stalled = new CountDownLatch(1); + CountDownLatch resume = new CountDownLatch(1); + var received = new ByteArrayOutputStream(); + try (var executor = Executors.newVirtualThreadPerTaskExecutor()) { + var writer = + executor.submit( + () -> { + response.getEntity() + .write( + new OutputStream() { + @Override + public void write(int value) { + received.write(value); + } + + @Override + public void write( + byte[] bytes, int offset, int length) + throws IOException { + stalled.countDown(); + try { + if (!resume.await( + 10, TimeUnit.SECONDS)) { + throw new IOException( + "Stalled writer was not released"); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException(e); + } + received.write(bytes, offset, length); + } + }); + return null; + }); + try { + assertTrue(stalled.await(10, TimeUnit.SECONDS)); + assertTrue(backpressured.await(10, TimeUnit.SECONDS)); + assertFalse(observerRef.get().isReady()); + // Each queued response contains at least 4096 payload bytes. + assertTrue(produced.get() * 4096 <= queueBudget); + } finally { + resume.countDown(); + } + writer.get(10, TimeUnit.SECONDS); + } + var input = new java.io.DataInputStream(new ByteArrayInputStream(received.toByteArray())); + for (int i = 0; i < 100; i++) { + assertEquals(0, input.readUnsignedByte()); + int size = input.readInt(); + assertEquals( + i + ":" + "x".repeat(4096), + StringValue.parseFrom(input.readNBytes(size)).getValue()); + } + assertEquals(-1, input.read()); + assertEquals("0", response.getTrailers().getFirstValue("grpc-status")); + response.notifyDelivered(null); + } + + @Test + void producerIgnoringReadinessFailsAtQueueBudget() throws Exception { + adapter.setMaxQueuedBytes(16384); + CountDownLatch produced = new CountDownLatch(1); + adapter.getServices() + .add( + ServerServiceDefinition.builder("review.Server") + .addMethod( + WATCH, + ServerCalls.asyncServerStreamingCall( + (StringValue request, + io.grpc.stub.StreamObserver + out) -> { + for (int i = 0; i < 16; i++) { + out.onNext( + StringValue.of("x".repeat(4096))); + } + out.onCompleted(); + produced.countDown(); + })) + .build()); + adapter.start(); + Response response = dispatch(WATCH, "watch"); + assertTrue(produced.await(10, TimeUnit.SECONDS)); + var bytes = new ByteArrayOutputStream(); + response.getEntity().write(bytes); + assertEquals(0, bytes.size()); + assertEquals("8", response.getTrailers().getFirstValue("grpc-status")); + assertEquals( + "Response queue byte budget exceeded", + response.getTrailers().getFirstValue("grpc-message")); + } + + @Test + void serverDeadlineCancelsHandlerAndReleasesAdmission() throws Exception { + adapter.setMaxConcurrentCalls(1); + adapter.setMaxCallDuration(Duration.ofMillis(200)); + CountDownLatch cancelled = new CountDownLatch(1); + start( + ServerServiceDefinition.builder("review.Server") + .addMethod( + ECHO, + ServerCalls.asyncUnaryCall( + (StringValue request, + io.grpc.stub.StreamObserver out) -> { + if ("block".equals(request.getValue())) { + ((ServerCallStreamObserver) out) + .setOnCancelHandler(cancelled::countDown); + } else { + out.onNext(request); + out.onCompleted(); + } + })) + .build()); + assertEquals( + Status.Code.DEADLINE_EXCEEDED, + assertThrows(StatusRuntimeException.class, () -> call("block")) + .getStatus() + .getCode()); + assertTrue(cancelled.await(10, TimeUnit.SECONDS)); + assertEquals("next", call("next")); + } + + @Test + void oversizedResponseFailsAndNextCallSucceeds() throws Exception { + adapter.setMaxMessageBytes(128); + start( + ServerServiceDefinition.builder("review.Server") + .addMethod( + ECHO, + ServerCalls.asyncUnaryCall( + (StringValue request, + io.grpc.stub.StreamObserver out) -> { + out.onNext( + "large".equals(request.getValue()) + ? StringValue.of("x".repeat(127)) + : request); + out.onCompleted(); + })) + .build()); + assertEquals( + Status.Code.RESOURCE_EXHAUSTED, + assertThrows(StatusRuntimeException.class, () -> call("large")) + .getStatus() + .getCode()); + assertEquals("next", call("next")); + } + + @Test + void shutdownWakesActiveStreamWriterAndDeliversCancellation() throws Exception { + CountDownLatch cancelled = new CountDownLatch(1); + adapter.getServices() + .add( + ServerServiceDefinition.builder("review.Server") + .addMethod( + WATCH, + ServerCalls.asyncServerStreamingCall( + (StringValue request, + io.grpc.stub.StreamObserver + out) -> { + ((ServerCallStreamObserver) out) + .setOnCancelHandler( + cancelled::countDown); + out.onNext(StringValue.of("first")); + })) + .build()); + adapter.start(); + Response response = dispatch(WATCH, "watch"); + CountDownLatch writing = new CountDownLatch(1); + try (var executor = Executors.newVirtualThreadPerTaskExecutor()) { + var writer = + executor.submit( + () -> { + response.getEntity() + .write( + new OutputStream() { + @Override + public void write(int value) { + writing.countDown(); + } + }); + return null; + }); + assertTrue(writing.await(10, TimeUnit.SECONDS)); + adapter.stop(); + writer.get(10, TimeUnit.SECONDS); + assertTrue(cancelled.await(10, TimeUnit.SECONDS)); + assertEquals("14", response.getTrailers().getFirstValue("grpc-status")); + assertFalse(adapter.isStarted()); + } + } + + @Test + void ownedCallbackShutdownTimeoutIsReportedAndInterruptsWorker() throws Exception { + adapter.setShutdownTimeout(Duration.ofMillis(200)); + CountDownLatch entered = new CountDownLatch(1); + CountDownLatch interrupted = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + start( + ServerServiceDefinition.builder("review.Server") + .addMethod( + ECHO, + ServerCalls.asyncUnaryCall( + (StringValue request, + io.grpc.stub.StreamObserver out) -> { + entered.countDown(); + try { + release.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + interrupted.countDown(); + } + })) + .build()); + ClientCalls.futureUnaryCall( + channel.newCall(ECHO, CallOptions.DEFAULT), StringValue.of("block")); + assertTrue(entered.await(10, TimeUnit.SECONDS)); + try { + assertEquals( + "gRPC callback executor shutdown timed out", + assertThrows(IllegalStateException.class, adapter::stop).getMessage()); + assertTrue(interrupted.await(10, TimeUnit.SECONDS)); + assertTrue(http.isStarted()); + assertFalse(adapter.isStarted()); + } finally { + release.countDown(); + } + } + + @Test + void clientCancellationAfterStreamMessageReleasesAdmission() throws Exception { + adapter.setMaxConcurrentCalls(1); + CountDownLatch cancelled = new CountDownLatch(1); + var service = ServerServiceDefinition.builder("review.Server"); + echo().getMethods().forEach(service::addMethod); + start( + service.addMethod( + WATCH, + ServerCalls.asyncServerStreamingCall( + (StringValue request, + io.grpc.stub.StreamObserver out) -> { + ((ServerCallStreamObserver) out) + .setOnCancelHandler(cancelled::countDown); + out.onNext(StringValue.of("first")); + })) + .build()); + CountDownLatch closed = new CountDownLatch(1); + AtomicReference status = new AtomicReference<>(); + var call = + channel.newCall(WATCH, CallOptions.DEFAULT.withDeadlineAfter(10, TimeUnit.SECONDS)); + call.start( + new ClientCall.Listener<>() { + @Override + public void onMessage(StringValue value) { + call.cancel("cancel after first message", null); + } + + @Override + public void onClose(Status value, Metadata trailers) { + status.set(value); + closed.countDown(); + } + }, + new Metadata()); + call.request(1); + call.sendMessage(StringValue.of("watch")); + call.halfClose(); + assertTrue(closed.await(10, TimeUnit.SECONDS)); + assertEquals(Status.Code.CANCELLED, status.get().getCode()); + assertTrue(cancelled.await(10, TimeUnit.SECONDS)); + assertEquals("next", call("next")); + } + + @Test + void disconnectRacingResponseSentTerminatesEachTransportOnce() throws Exception { + adapter.setMaxConcurrentCalls(1); + adapter.getServices() + .add( + ServerServiceDefinition.builder("review.Server") + .addMethod( + WATCH, + ServerCalls.asyncServerStreamingCall( + (StringValue request, + io.grpc.stub.StreamObserver + out) -> out.onNext(request))) + .build()); + adapter.start(); + try (var executor = Executors.newVirtualThreadPerTaskExecutor()) { + for (int i = 0; i < 32; i++) { + Response response = dispatch(WATCH, "watch"); + assertEquals(org.restlet.data.Status.SUCCESS_OK, response.getStatus()); + var writer = + executor.submit( + () -> { + response.getEntity().write(OutputStream.nullOutputStream()); + response.notifyDelivered(null); + return null; + }); + var disconnect = + executor.submit( + () -> + response.getRequest() + .notifyDisconnect(new IOException("Peer reset"))); + disconnect.get(10, TimeUnit.SECONDS); + writer.get(10, TimeUnit.SECONDS); + } + } + adapter.stop(); + } +} diff --git a/org.restlet/src/main/java/org/restlet/Response.java b/org.restlet/src/main/java/org/restlet/Response.java index ed4c669641..373c80eb75 100644 --- a/org.restlet/src/main/java/org/restlet/Response.java +++ b/org.restlet/src/main/java/org/restlet/Response.java @@ -12,6 +12,8 @@ import java.util.Date; import java.util.List; import java.util.Set; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionStage; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CopyOnWriteArraySet; import org.restlet.data.AuthenticationInfo; @@ -37,6 +39,9 @@ public class Response extends Message { private static final ThreadLocal CURRENT = new ThreadLocal<>(); + /** Completes once the server connector has finished writing this response. */ + private final CompletableFuture deliverySignal = new CompletableFuture<>(); + /** * Returns the response associated with the current thread. * @@ -154,6 +159,29 @@ public static void setCurrent(Response response) { * * @param request The request associated with this response. */ + /** + * Returns a read-only completion stage for the delivery of this response by the server + * connector. It completes once, when the terminal write is done: with null if it succeeded, or + * with the failure. Dependent actions may run on a connector I/O thread, so they must be short + * and must not block; the "onSent" callback is the place for other work. Not all connectors + * support this signal. + * + * @return The delivery signal. + */ + public CompletionStage getDeliverySignal() { + return deliverySignal.minimalCompletionStage(); + } + + /** + * Reports the end of the delivery of this response; called by server connectors. The first + * report wins. + * + * @param failure The failure of the terminal write, or null if the response was delivered. + */ + public void notifyDelivered(Throwable failure) { + deliverySignal.complete(failure); + } + public Response(Request request) { this.age = 0; this.accessControlAllowCredentials = null; diff --git a/org.restlet/src/main/java/org/restlet/engine/Engine.java b/org.restlet/src/main/java/org/restlet/engine/Engine.java index 262aaf74fa..dbef4e29be 100644 --- a/org.restlet/src/main/java/org/restlet/engine/Engine.java +++ b/org.restlet/src/main/java/org/restlet/engine/Engine.java @@ -119,29 +119,37 @@ public static synchronized void clear() { * @return The thread with proper variables ready to run the given runnable task. */ public static Thread createThreadWithLocalVariables(final Runnable runnable, String name) { + return new Thread(withLocalVariables(runnable), name); + } + + /** + * Returns a task running the given one with the thread local variables of the Restlet API and + * engine as they are now, in the calling thread, then clearing them. + * + * @param runnable The task to run. + * @return The task restoring the current thread local variables. + */ + public static Runnable withLocalVariables(final Runnable runnable) { // Save the thread local variables final org.restlet.Application currentApplication = org.restlet.Application.getCurrent(); final Context currentContext = Context.getCurrent(); final Integer currentVirtualHost = org.restlet.routing.VirtualHost.getCurrent(); final Response currentResponse = Response.getCurrent(); - Runnable r = - () -> { - // Copy the thread local variables - Response.setCurrent(currentResponse); - Context.setCurrent(currentContext); - org.restlet.routing.VirtualHost.setCurrent(currentVirtualHost); - org.restlet.Application.setCurrent(currentApplication); - - try { - // Run the user task - runnable.run(); - } finally { - Engine.clearThreadLocalVariables(); - } - }; - - return new Thread(r, name); + return () -> { + // Copy the thread local variables + Response.setCurrent(currentResponse); + Context.setCurrent(currentContext); + org.restlet.routing.VirtualHost.setCurrent(currentVirtualHost); + org.restlet.Application.setCurrent(currentApplication); + + try { + // Run the user task + runnable.run(); + } finally { + Engine.clearThreadLocalVariables(); + } + }; } /** Clears the thread local variables set by the Restlet API and engine. */ diff --git a/org.restlet/src/main/java/org/restlet/engine/header/HeaderConstants.java b/org.restlet/src/main/java/org/restlet/engine/header/HeaderConstants.java index ad65800621..928de2d3e8 100644 --- a/org.restlet/src/main/java/org/restlet/engine/header/HeaderConstants.java +++ b/org.restlet/src/main/java/org/restlet/engine/header/HeaderConstants.java @@ -200,6 +200,16 @@ private HeaderConstants() { public static final String ATTRIBUTE_VERSION = "org.restlet.http.version"; + /** + * Response attribute holding a {@code Series

} that server connectors supporting it + * write as-is, each name replacing the values produced from the Restlet API. Headers defining + * the message framing, such as Content-Type, Content-Length or Connection, are ignored. Unlike + * {@link org.restlet.Message#getHeaders()}, it may carry standard headers, for protocols + * relaying headers they don't interpret. + */ + public static final String ATTRIBUTE_RAW_RESPONSE_HEADERS = + "org.restlet.http.rawResponseHeaders"; + public static final String ATTRIBUTE_HTTPS_KEY_SIZE = "org.restlet.https.keySize"; public static final String ATTRIBUTE_HTTPS_SSL_SESSION_ID = "org.restlet.https.sslSessionId"; diff --git a/org.restlet/src/main/java/org/restlet/engine/netty/NettyRequest.java b/org.restlet/src/main/java/org/restlet/engine/netty/NettyRequest.java index 9a6a442f6f..b250fc31f0 100644 --- a/org.restlet/src/main/java/org/restlet/engine/netty/NettyRequest.java +++ b/org.restlet/src/main/java/org/restlet/engine/netty/NettyRequest.java @@ -109,16 +109,10 @@ class NettyRequest extends Request { this.headers = call.getRequestHeaders(); setMethod(Method.valueOf(call.getMethod())); + setProtocol(Protocol.valueOf(call.getProtocol().getSchemeName(), call.getVersion())); setHostRef(hostRef(call.getProtocol(), headers)); - String uri = call.getRequestUri(); - Reference resourceRef = new Reference(getHostRef(), uri); - if (resourceRef.isRelative()) { - // Take care of the "/" between the host part and the segments. - resourceRef = - new Reference(getHostRef().toString() + (uri.startsWith("/") ? "" : "/") + uri); - } - setResourceRef(resourceRef); + setResourceRef(resourceRef(call.getRequestUri(), getHostRef())); setOriginalRef(ReferenceUtils.getOriginalRef(getResourceRef(), headers)); String dateHeader = headers.getFirstValue(HeaderConstants.HEADER_DATE, true); @@ -131,6 +125,55 @@ public Series
getTrailers() { return call.getRequestTrailers(); } + /** + * Returns the resource reference of a request target, relative to the host reference. An + * absolute target for the same authority is reduced to its path and query; another absolute + * target is kept as is. + */ + private static Reference resourceRef(String uri, Reference hostRef) { + String base = hostRef.toString(); + String target = uri; + if (target == null || target.isEmpty()) { + target = "/"; + } else if (target.regionMatches(true, 0, "http://", 0, 7) + || target.regionMatches(true, 0, "https://", 0, 8)) { + Reference absoluteRef = new Reference(target); + if (!sameAuthority(absoluteRef, hostRef)) { + absoluteRef.setBaseRef(new Reference(base)); + return absoluteRef; + } + String path = absoluteRef.getPath(); + target = (path == null || path.isEmpty()) ? "/" : path; + if (!target.startsWith("/")) { + target = "/" + target; + } + if (absoluteRef.getQuery() != null) { + target += "?" + absoluteRef.getQuery(); + } + } else if (!target.startsWith("/")) { + target = "/" + target; + } + Reference result = new Reference(base + target); + result.setBaseRef(new Reference(base)); + return result; + } + + /** Compares the scheme, host (ignoring case) and effective port of two references. */ + private static boolean sameAuthority(Reference target, Reference hostRef) { + String host = target.getHostDomain(); + return host != null + && hostRef.getHostDomain() != null + && target.getSchemeProtocol() != null + && target.getSchemeProtocol().equals(hostRef.getSchemeProtocol()) + && host.equalsIgnoreCase(hostRef.getHostDomain()) + && effectivePort(target) == effectivePort(hostRef); + } + + private static int effectivePort(Reference reference) { + int port = reference.getHostPort(); + return port == -1 ? reference.getSchemeProtocol().getDefaultPort() : port; + } + /** Returns the host reference built from the "Host" header. */ private static String hostRef(Protocol protocol, Series
headers) { StringBuilder sb = new StringBuilder(); diff --git a/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerCall.java b/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerCall.java index 717164abde..299392f227 100644 --- a/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerCall.java +++ b/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerCall.java @@ -19,6 +19,7 @@ import io.netty.channel.ChannelFutureListener; import io.netty.channel.ChannelHandler; import io.netty.channel.ChannelPipeline; +import io.netty.channel.ChannelPromise; import io.netty.handler.codec.http.DefaultFullHttpResponse; import io.netty.handler.codec.http.DefaultHttpContent; import io.netty.handler.codec.http.DefaultHttpResponse; @@ -48,12 +49,14 @@ import java.util.List; import java.util.Locale; import java.util.Map; +import java.util.Set; import java.util.logging.Level; import java.util.logging.Logger; import javax.net.ssl.SSLPeerUnverifiedException; import javax.net.ssl.SSLSession; import org.restlet.Context; import org.restlet.Response; +import org.restlet.Uniform; import org.restlet.data.Header; import org.restlet.data.MediaType; import org.restlet.data.Method; @@ -113,6 +116,23 @@ class NettyServerCall { /** The response headers. */ private final Series
responseHeaders = new Series<>(Header.class); + /** Raw response headers that would break the message framing. */ + private static final Set FRAMING_HEADERS = + Set.of( + "content-type", + "content-length", + "content-encoding", + "transfer-encoding", + "connection", + "keep-alive", + "proxy-connection", + "upgrade", + "te", + "trailer"); + + /** A failure writing the response body that left the connection open, if any. */ + private volatile Throwable bodyFailure; + /** The trailing headers received with the request entity. */ private final Series
requestTrailers = new Series<>(Header.class); @@ -301,6 +321,12 @@ String getMethod() { /** The request target, "/" when empty. */ String getRequestUri() { String uri = request.uri(); + if (uri == null || uri.isEmpty() || "/".equals(uri)) { + String path = request.headers().get(":path"); + if (path != null && !path.isEmpty()) { + uri = path; + } + } return (uri == null || uri.isEmpty()) ? "/" : uri; } @@ -328,6 +354,16 @@ Protocol getProtocol() { return confidential ? Protocol.HTTPS : Protocol.HTTP; } + /** The protocol version as reported to the application (for example "1.1" or "2.0"). */ + String getVersion() { + String version = + kind == StreamKind.HTTP1 ? request.protocolVersion().text() : kind.version(); + int slash = version.lastIndexOf('/'); + return (slash >= 0 && slash + 1 < version.length()) + ? version.substring(slash + 1) + : version; + } + /** The remote address, or null. */ InetSocketAddress getRemoteAddress() { SocketAddress address = @@ -426,9 +462,78 @@ void commit(Response response) { } catch (IOException | RuntimeException e) { handleException(response, e); } finally { - complete(); - if (response.getOnSent() != null) { - response.getOnSent().handle(response.getRequest(), response); + // Delivery is reported by the terminal write itself, before the peer can see it. + notifySent(response, complete(f -> response.notifyDelivered(deliveryFailure(f)))); + } + } + + /** + * Reports the delivery of the response once its terminal write is done, then invokes its + * "onSent" callback, never on the event loop: in the calling thread if the write is already + * done, else in a virtual thread. If the write failed, the response status is first set to + * {@link Status#CONNECTOR_ERROR_COMMUNICATION} with the cause, since the response wasn't + * delivered. + * + * @param response The response. + * @param delivery The terminal write. + */ + private void notifySent(Response response, ChannelFuture delivery) { + Uniform onSent = response.getOnSent(); + if (onSent == null) { + return; + } + if (delivery.isDone() && !channel.eventLoop().inEventLoop()) { + // Done doesn't mean its listeners ran: the delivery is reported first, once. + Throwable failure = deliveryFailure(delivery); + response.notifyDelivered(failure); + sent(onSent, response, failure); + } else { + // The thread local variables of the call are captured here, on its worker. + var failure = new java.util.concurrent.atomic.AtomicReference(); + Runnable task = Engine.withLocalVariables(() -> sent(onSent, response, failure.get())); + delivery.addListener( + f -> { + failure.set(deliveryFailure(f)); + Thread.ofVirtual().name("restlet-response-sent").start(task); + }); + } + } + + /** Returns the failure of the body or terminal write, or null if the response was sent. */ + private Throwable deliveryFailure(io.netty.util.concurrent.Future delivery) { + Throwable failure = bodyFailure; + if (failure == null && !delivery.isSuccess()) { + failure = delivery.cause(); + } + return failure; + } + + private static void sent(Uniform onSent, Response response, Throwable failure) { + if (failure != null) { + response.setStatus(new Status(Status.CONNECTOR_ERROR_COMMUNICATION, failure)); + } + onSent.handle(response.getRequest(), response); + } + + /** + * Writes the headers of the {@link HeaderConstants#ATTRIBUTE_RAW_RESPONSE_HEADERS} attribute, + * each name replacing the values produced from the Restlet API. + */ + private void addRawHeaders(Response response) { + if (!(response.getAttributes().get(HeaderConstants.ATTRIBUTE_RAW_RESPONSE_HEADERS) + instanceof Series raw)) { + return; + } + Set replaced = new java.util.HashSet<>(); + for (Object item : raw) { + if (item instanceof Header header) { + String name = header.getName().toLowerCase(Locale.ROOT); + if (!FRAMING_HEADERS.contains(name)) { + if (replaced.add(name)) { + responseHeaders.removeAll(header.getName(), true); + } + responseHeaders.add(header.getName(), header.getValue()); + } } } } @@ -536,6 +641,7 @@ private void addResponseHeaders(Response response) { try { HeaderUtils.addGeneralHeaders(response, responseHeaders); HeaderUtils.addResponseHeaders(response, responseHeaders); + addRawHeaders(response); if (response.getStatus() != null) { setStatus(response.getStatus().getCode(), response.getStatus().getReasonPhrase()); } @@ -554,6 +660,7 @@ private void setStatus(int code, String reason) { private void handleException(Response response, Exception exception) { if (isConnectionBroken(exception)) { + bodyFailure = exception; getLogger() .log( Level.INFO, @@ -737,10 +844,24 @@ private boolean isRequestBodyDiscardable() { return length >= 0 && length - requestBody.getReceivedBytes() <= MAX_DISCARDED_BYTES; } - /** Completes the exchange: ends the response and frees the connection for the next request. */ - void complete() { + /** + * Completes the exchange: ends the response and frees the connection for the next request. + * + * @return The terminal write of the response, failed if the connection is closed. + */ + ChannelFuture complete() { + return complete(null); + } + + /** + * Completes the exchange: ends the response and frees the connection for the next request. + * + * @param delivery A listener of the terminal write, added before it is submitted, or null. + * @return The terminal write of the response, failed if the connection is closed. + */ + ChannelFuture complete(ChannelFutureListener delivery) { if (completed) { - return; + return listen(channel.newSucceededFuture(), delivery); } completed = true; @@ -758,13 +879,13 @@ void complete() { if (!channel.isActive()) { requestBody.close(); - return; + return listen(channel.newFailedFuture(new ClosedChannelException()), delivery); } if (responseEndedWithHead) { - trackDelivery(responseEndFuture); + listen(trackDelivery(responseEndFuture), delivery); channel.flush(); requestBody.close(); - return; + return responseEndFuture; } // The response head announced whether the connection persists. If so, the unread @@ -777,7 +898,9 @@ void complete() { copyHeaders(responseTrailers, withTrailers.trailingHeaders(), kind.isStream()); lastContent = withTrailers; } - ChannelFuture future = trackDelivery(channel.writeAndFlush(lastContent)); + ChannelPromise future = channel.newPromise(); + listen(trackDelivery(future), delivery); + channel.writeAndFlush(lastContent, future); if (keepAlive) { requestBody.discard(); } else { @@ -787,6 +910,7 @@ void complete() { if (!kind.isStream()) { completeHttp1(future, keepAlive); } + return future; } finally { if (eventStreamPermit) { channel.attr(NettyServerHandler.EVENT_STREAM_ACTIVE).set(false); @@ -815,6 +939,10 @@ private ChannelFuture trackDelivery(ChannelFuture future) { return future; } + private static ChannelFuture listen(ChannelFuture future, ChannelFutureListener listener) { + return listener == null ? future : future.addListener(listener); + } + private synchronized void markResponseDelivered() { responseDelivered = true; } @@ -860,12 +988,13 @@ private static void copyHeaders( } /** - * Aborts the exchange by closing the connection. + * Aborts the exchange: resets its stream for HTTP/2 and HTTP/3, leaving the other streams of + * the connection, or closes the HTTP/1.1 connection. * * @return True. */ boolean abort() { - connection.close(); + channel.close(); return true; } diff --git a/org.restlet/src/main/java/org/restlet/util/WrapperResponse.java b/org.restlet/src/main/java/org/restlet/util/WrapperResponse.java index 00fd5c1862..7cc6d16c30 100644 --- a/org.restlet/src/main/java/org/restlet/util/WrapperResponse.java +++ b/org.restlet/src/main/java/org/restlet/util/WrapperResponse.java @@ -38,6 +38,16 @@ public class WrapperResponse extends Response { /** The wrapped response. */ private final Response wrappedResponse; + @Override + public java.util.concurrent.CompletionStage getDeliverySignal() { + return wrappedResponse.getDeliverySignal(); + } + + @Override + public void notifyDelivered(Throwable failure) { + wrappedResponse.notifyDelivered(failure); + } + /** * Constructor. * diff --git a/org.restlet/src/test/java/org/restlet/ResponseDeliveryTestCase.java b/org.restlet/src/test/java/org/restlet/ResponseDeliveryTestCase.java new file mode 100644 index 0000000000..86097945a4 --- /dev/null +++ b/org.restlet/src/test/java/org/restlet/ResponseDeliveryTestCase.java @@ -0,0 +1,67 @@ +/** + * 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; + +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; + +import java.io.IOException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; +import org.junit.jupiter.api.Test; +import org.restlet.engine.Engine; +import org.restlet.util.WrapperResponse; + +class ResponseDeliveryTestCase { + + @Test + void firstDeliveryReportWins() throws Exception { + Response response = new Response(new Request()); + IOException failure = new IOException("Peer reset"); + response.notifyDelivered(failure); + response.notifyDelivered(null); + assertSame( + failure, + response.getDeliverySignal().toCompletableFuture().get(5, TimeUnit.SECONDS)); + } + + @Test + void wrapperSharesTheDeliverySignalOfItsResponse() throws Exception { + Response response = new Response(new Request()); + Response wrapper = new WrapperResponse(response); + response.notifyDelivered(null); + assertNull(wrapper.getDeliverySignal().toCompletableFuture().get(5, TimeUnit.SECONDS)); + + Response other = new Response(new Request()); + new WrapperResponse(other).notifyDelivered(null); + assertNull(other.getDeliverySignal().toCompletableFuture().get(5, TimeUnit.SECONDS)); + } + + @Test + void localVariablesAreRestoredInAnotherThreadThenCleared() throws Exception { + Response response = new Response(new Request()); + Response.setCurrent(response); + AtomicReference seen = new AtomicReference<>(); + AtomicReference after = new AtomicReference<>(response); + try { + Runnable task = Engine.withLocalVariables(() -> seen.set(Response.getCurrent())); + Thread.ofVirtual() + .start( + () -> { + task.run(); + after.set(Response.getCurrent()); + }) + .join(); + } finally { + Engine.clearThreadLocalVariables(); + } + assertSame(response, seen.get()); + assertNull(after.get()); + } +} diff --git a/org.restlet/src/test/java/org/restlet/engine/netty/NettyRequestMappingTestCase.java b/org.restlet/src/test/java/org/restlet/engine/netty/NettyRequestMappingTestCase.java new file mode 100644 index 0000000000..d220b936c4 --- /dev/null +++ b/org.restlet/src/test/java/org/restlet/engine/netty/NettyRequestMappingTestCase.java @@ -0,0 +1,133 @@ +/** + * 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.engine.netty; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertSame; + +import io.netty.channel.embedded.EmbeddedChannel; +import io.netty.handler.codec.http.DefaultHttpRequest; +import io.netty.handler.codec.http.HttpMethod; +import io.netty.handler.codec.http.HttpVersion; +import org.junit.jupiter.api.Test; +import org.restlet.Context; +import org.restlet.Response; +import org.restlet.Restlet; +import org.restlet.routing.Router; +import org.restlet.routing.Template; + +class NettyRequestMappingTestCase { + @Test + void serviceRoutesMatchOriginAndSameHostAbsoluteTargetsForEveryHttpVersion() { + for (StreamKind kind : StreamKind.values()) { + for (String target : + new String[] { + "/review.Server/Echo?mode=one", + "review.Server/Echo?mode=one", + "http://localhost:8080/review.Server/Echo?mode=one" + }) { + EmbeddedChannel channel = new EmbeddedChannel(); + try { + var head = + new DefaultHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.POST, target); + head.headers().set("host", "localhost:8080"); + var call = + new NettyServerCall( + null, + channel, + kind, + head, + new ChannelInputStream(channel, 1000), + false); + var request = call.toRequest(new Context()); + assertEquals("/review.Server/Echo", request.getResourceRef().getPath()); + assertEquals("mode=one", request.getResourceRef().getQuery()); + assertEquals(kind.version().substring(5), request.getProtocol().getVersion()); + Router router = new Router(); + Restlet service = new Restlet() {}; + router.attach("/review.Server/", service, Template.MODE_STARTS_WITH); + assertSame( + router.getRoutes().get(0), + router.getNext(request, new Response(request))); + assertEquals("/review.Server/Echo", request.getOriginalRef().getPath()); + } finally { + channel.finishAndReleaseAll(); + } + } + } + } + + private static org.restlet.Request map( + StreamKind kind, HttpVersion version, String target, String host) { + EmbeddedChannel channel = new EmbeddedChannel(); + try { + var head = new DefaultHttpRequest(version, HttpMethod.GET, target); + head.headers().set("host", host); + return new NettyServerCall( + null, channel, kind, head, new ChannelInputStream(channel, 1000), false) + .toRequest(new Context()); + } finally { + channel.finishAndReleaseAll(); + } + } + + @Test + void absoluteTargetsWithAnEquivalentAuthorityMapToTheirPath() { + for (String target : + new String[] { + "http://example.com:80/rest?q=1", + "http://EXAMPLE.com/rest?q=1", + "HTTP://example.com/rest?q=1" + }) { + var request = map(StreamKind.HTTP2, HttpVersion.HTTP_1_1, target, "example.com:80"); + assertEquals("http://example.com/rest?q=1", request.getResourceRef().toString()); + assertEquals("/rest", request.getResourceRef().getRemainingPart(false, false)); + } + } + + @Test + void absoluteTargetsForAnotherAuthorityAreKept() { + var request = + map(StreamKind.HTTP1, HttpVersion.HTTP_1_1, "http://other.test:8080/rest", "a"); + assertEquals("http://other.test:8080/rest", request.getResourceRef().toString()); + assertEquals("http://a", request.getResourceRef().getBaseRef().toString()); + } + + @Test + void http1ReportsTheVersionOfItsRequestLine() { + assertEquals( + "1.0", + map(StreamKind.HTTP1, HttpVersion.HTTP_1_0, "/", "a").getProtocol().getVersion()); + assertEquals( + "2.0", + map(StreamKind.HTTP2, HttpVersion.HTTP_1_1, "/", "a").getProtocol().getVersion()); + } + + @Test + void emptyTargetMapsToRoot() { + EmbeddedChannel channel = new EmbeddedChannel(); + try { + var head = new DefaultHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET, ""); + head.headers().set("host", "localhost"); + var call = + new NettyServerCall( + null, + channel, + StreamKind.HTTP1, + head, + new ChannelInputStream(channel, 1000), + false); + var request = call.toRequest(new Context()); + assertEquals("/", request.getResourceRef().getPath()); + assertEquals("/", request.getResourceRef().getRemainingPart(false, false)); + } finally { + channel.finishAndReleaseAll(); + } + } +}