Fix flaky Netty connector tests and Sonar issue from #1553

org.restlet/src/main/java/org/restlet/engine/netty/ChannelOutputStream.java

@@ -12,7 +12,12 @@
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelFuture;
 import java.io.IOException;
+import java.io.InterruptedIOException;
 import java.io.OutputStream;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.Function;
 
 /**
@@ -42,6 +47,16 @@ final class ChannelOutputStream extends OutputStream {
 
     private boolean closed;
 
+    /** Bytes written to the channel whose write has not completed yet. */
+    private final AtomicLong inFlightBytes = new AtomicLong();
+
+    private final ReentrantLock inFlightLock = new ReentrantLock();
+
+    private final Condition inFlightChanged = inFlightLock.newCondition();
+
+    /** The cause of the first failed write, if any. */
+    private volatile Throwable writeFailure;
+
     /**
      * Constructor.
      *
@@ -75,8 +90,8 @@ public void write(int b) throws IOException {
 
     @Override
     public void write(byte[] b, int off, int len) throws IOException {
-        ensureOpen();
         while (len > 0) {
+            ensureOpen();
             ensureBuffer();
             int n = Math.min(len, buffer.writableBytes());
             buffer.writeBytes(b, off, n);
@@ -131,19 +146,88 @@ private void ensureOpen() throws IOException {
         if (closed) {
             throw new IOException("Stream closed");
         }
+        checkWriteFailure();
         if (!channel.isActive()) {
             throw new IOException("Connection closed by peer");
         }
     }
 
+    private void checkWriteFailure() throws IOException {
+        Throwable cause = writeFailure;
+        if (cause != null) {
+            throw new IOException("Unable to write to the connection", cause);
+        }
+    }
+
     private void send(boolean flush) throws IOException {
         Writability.await(channel, writeTimeoutMs);
+        awaitInFlightBytes();
         ByteBuf chunk = buffer;
         buffer = null;
+        int size = chunk.readableBytes();
+        inFlightBytes.addAndGet(size);
         Object msg = contentFactory.apply(chunk);
         ChannelFuture future = flush ? channel.writeAndFlush(msg) : channel.write(msg);
+        future.addListener(f -> onWriteDone(size, f.cause()));
         if (future.isDone() && !future.isSuccess()) {
             throw new IOException("Unable to write to the connection", future.cause());
         }
     }
+
+    /** Called when a write completes, possibly on the event loop. */
+    private void onWriteDone(int size, Throwable cause) {
+        inFlightLock.lock();
+        try {
+            inFlightBytes.addAndGet(-size);
+            if (cause != null && writeFailure == null) {
+                writeFailure = cause;
+            }
+            inFlightChanged.signalAll();
+        } finally {
+            inFlightLock.unlock();
+        }
+    }
+
+    /**
+     * Waits until the bytes written but not yet sent fall below the channel high water mark. Writes
+     * made outside the event loop only update the channel writability once the event loop has
+     * processed them, so a fast producer could otherwise queue unbounded data, and miss the failure
+     * of writes to a connection closed in the meantime.
+     */
+    private void awaitInFlightBytes() throws IOException {
+        int limit = Math.max(bufferSize, channel.config().getWriteBufferHighWaterMark());
+        if (inFlightBytes.get() <= limit || channel.eventLoop().inEventLoop()) {
+            checkWriteFailure();
+            return;
+        }
+
+        channel.flush();
+        long remaining =
+                (writeTimeoutMs > 0)
+                        ? TimeUnit.MILLISECONDS.toNanos(writeTimeoutMs)
+                        : Long.MAX_VALUE;
+        inFlightLock.lock();
+        try {
+            while (inFlightBytes.get() > limit && writeFailure == null) {
+                if (!channel.isActive()) {
+                    throw new IOException("Connection closed by peer");
+                }
+                if (remaining <= 0) {
+                    throw new InterruptedIOException("Write timed out");
+                }
+                // Wake up periodically in case the channel closes without failing a write.
+                long slice = Math.min(remaining, TimeUnit.MILLISECONDS.toNanos(100));
+                long left = inFlightChanged.awaitNanos(slice);
+                if (writeTimeoutMs > 0) {
+                    remaining -= (slice - Math.max(0, left));
+                }
+            }
+        } catch (InterruptedException _) {
+            Thread.currentThread().interrupt();
+            throw new InterruptedIOException("Interrupted while writing");
+        } finally {
+            inFlightLock.unlock();
+        }
+        checkWriteFailure();
+    }
 }

org.restlet/src/main/java/org/restlet/engine/netty/NettyServerCall.java

@@ -608,6 +608,27 @@ private void writeHead(boolean emptyBody) {
         responseHeadWritten = true;
 
         int code = statusCode;
+        HttpResponse head = newResponseHead(code);
+        boolean bodyAllowed =
+                code != 204 && code != 304 && code >= 200 && !"HEAD".equals(getMethod());
+        boolean hasTrailers = bodyAllowed && !responseTrailers.isEmpty();
+        // Streams can carry trailers added after the head was written.
+        responseTrailersEnabled = hasTrailers || (kind.isStream() && bodyAllowed);
+
+        if (kind.isStream()) {
+            // Connection-specific headers are forbidden in HTTP/2 and HTTP/3.
+            head.headers().remove(HttpHeaderNames.CONNECTION);
+            head.headers().remove(HttpHeaderNames.TRANSFER_ENCODING);
+            head.headers().remove("Keep-Alive");
+        } else {
+            setHttp1Framing(head, emptyBody, hasTrailers, bodyAllowed);
+        }
+
+        channel.write(head);
+    }
+
+    /** Creates the response head with the status and the response headers. */
+    private HttpResponse newResponseHead(int code) {
         String reason = reasonPhrase;
         HttpResponseStatus status =
                 (reason != null && !reason.isEmpty())
@@ -624,45 +645,27 @@ private void writeHead(boolean emptyBody) {
         if (altSvc != null && !head.headers().contains("Alt-Svc")) {
             head.headers().set(kind.isStream() ? "alt-svc" : "Alt-Svc", altSvc);
         }
-
-        boolean hasTrailers = !responseTrailers.isEmpty();
-        boolean hasLength = head.headers().contains(HttpHeaderNames.CONTENT_LENGTH);
-        boolean noBodyStatus = code == 204 || code == 304 || code < 200;
-
-        if (noBodyStatus || "HEAD".equals(getMethod())) {
-            hasTrailers = false;
-        }
-        responseTrailersEnabled =
-                hasTrailers || (kind.isStream() && !noBodyStatus && !"HEAD".equals(getMethod()));
-
-        if (kind.isStream()) {
-            // Connection-specific headers are forbidden in HTTP/2 and HTTP/3.
-            head.headers().remove(HttpHeaderNames.CONNECTION);
-            head.headers().remove(HttpHeaderNames.TRANSFER_ENCODING);
-            head.headers().remove("Keep-Alive");
-        } else {
-            if (emptyBody) {
-                if (hasTrailers) {
-                    head.headers().remove(HttpHeaderNames.CONTENT_LENGTH);
-                    HttpUtil.setTransferEncodingChunked(head, true);
-                } else if (!hasLength && !noBodyStatus && !"HEAD".equals(getMethod())) {
-                    head.headers().set(HttpHeaderNames.CONTENT_LENGTH, 0);
-                }
-            } else if (hasTrailers) {
-                head.headers().remove(HttpHeaderNames.CONTENT_LENGTH);
-                HttpUtil.setTransferEncodingChunked(head, true);
-            } else if (!hasLength) {
-                head.headers().set(HttpHeaderNames.TRANSFER_ENCODING, HttpHeaderValues.CHUNKED);
-            }
-            if (hasTrailers) {
-                setTrailerNames(head.headers(), responseTrailers);
-            }
-            if (!HttpUtil.isKeepAlive(request)) {
-                head.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.CLOSE);
+        return head;
+    }
+
+    /** Sets the HTTP/1.1 message framing and connection headers of the response head. */
+    private void setHttp1Framing(
+            HttpResponse head, boolean emptyBody, boolean hasTrailers, boolean bodyAllowed) {
+        HttpHeaders headers = head.headers();
+        if (hasTrailers) {
+            headers.remove(HttpHeaderNames.CONTENT_LENGTH);
+            HttpUtil.setTransferEncodingChunked(head, true);
+            setTrailerNames(headers, responseTrailers);
+        } else if (!headers.contains(HttpHeaderNames.CONTENT_LENGTH)) {
+            if (!emptyBody) {
+                headers.set(HttpHeaderNames.TRANSFER_ENCODING, HttpHeaderValues.CHUNKED);
+            } else if (bodyAllowed) {
+                headers.set(HttpHeaderNames.CONTENT_LENGTH, 0);
             }
         }
-
-        channel.write(head);
+        if (!HttpUtil.isKeepAlive(request)) {
+            headers.set(HttpHeaderNames.CONNECTION, HttpHeaderValues.CLOSE);
+        }
     }
 
     /** Completes the exchange: ends the response and frees the connection for the next request. */

org.restlet/src/main/java/org/restlet/engine/netty/NettyServerHandler.java

@@ -115,6 +115,11 @@ private void onRequest(ChannelHandlerContext ctx, HttpRequest request) {
         ChannelInputStream body =
                 new ChannelInputStream(ctx.channel(), helper.getConnectorIdleTimeout());
         currentBody = body;
+        if (kind == StreamKind.HTTP1 && !hasBody(request)) {
+            // End the body now: the call could otherwise complete before the last (empty) content
+            // is delivered, and close a connection that the response announced as kept alive.
+            body.end();
+        }
 
         NettyServerCall call =
                 new NettyServerCall(helper, ctx.channel(), kind, request, body, confidential);
@@ -190,6 +195,12 @@ private void onContent(ChannelHandlerContext ctx, HttpContent content) {
         // Otherwise the body stream requests the next read once the consumer catches up.
     }
 
+    /** Indicates whether an HTTP/1.1 request has a body, according to its framing headers. */
+    private static boolean hasBody(HttpRequest request) {
+        return HttpUtil.isTransferEncodingChunked(request)
+                || HttpUtil.getContentLength(request, 0L) != 0;
+    }
+
     private void runCall(NettyServerCall call) {
         try {
             helper.handle(call);