Flush each full Netty output buffer

changes.md

@@ -231,6 +231,9 @@ Changes log
           "/mcp", "/mcp/tools" and "/mcp?x=1" but not "/mcpx". Useful for mounting several
           routers or applications under path prefixes on one connector.
     - Bugs fixed
+        - The Netty connectors now flush each full output buffer of a request or response body,
+          so a producer that pauses after filling one no longer holds back what it produced
+          until it resumes or ends. Issue #1632.
         - `Client.setContext()` now stores the connector helper in the "org.restlet.engine.helper"
           context attribute, as the constructors do, so clients added with
           `Component.getClients().add(Protocol)` expose it too.

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

@@ -21,8 +21,9 @@
 import java.util.function.Function;
 
 /**
- * Blocking output stream writing to a Netty channel. Bytes are buffered locally and written as
- * content messages when the buffer is full or on {@link #flush()}.
+ * Blocking output stream writing to a Netty channel. Bytes are buffered locally and sent as content
+ * messages when the buffer is full or on {@link #flush()}. A full buffer is flushed to the peer at
+ * once, so a producer that pauses after filling it doesn't hold back what it produced.
  *
  * <p>Back-pressure: before each write, the calling thread waits until the channel is writable, so a
  * slow peer slows down the producer instead of filling the memory.
@@ -84,7 +85,7 @@ public void write(int b) throws IOException {
         ensureBuffer();
         buffer.writeByte(b);
         if (buffer.writableBytes() == 0) {
-            send(false);
+            send();
         }
     }
 
@@ -98,7 +99,7 @@ public void write(byte[] b, int off, int len) throws IOException {
             off += n;
             len -= n;
             if (buffer.writableBytes() == 0) {
-                send(false);
+                send();
             }
         }
     }
@@ -107,7 +108,7 @@ public void write(byte[] b, int off, int len) throws IOException {
     public void flush() throws IOException {
         ensureOpen();
         if (buffer != null && buffer.isReadable()) {
-            send(true);
+            send();
         } else {
             channel.flush();
         }
@@ -124,7 +125,7 @@ public void close() throws IOException {
         if (!closed) {
             try {
                 if (buffer != null && buffer.isReadable()) {
-                    send(true);
+                    send();
                 }
             } finally {
                 closed = true;
@@ -159,15 +160,15 @@ private void checkWriteFailure() throws IOException {
         }
     }
 
-    private void send(boolean flush) throws IOException {
+    private void send() 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);
+        ChannelFuture future = channel.writeAndFlush(msg);
         future.addListener(f -> onWriteDone(size, f.cause()));
         if (future.isDone() && !future.isSuccess()) {
             throw new IOException("Unable to write to the connection", future.cause());

org.restlet/src/test/java/org/restlet/engine/connector/FullBufferFlushTestCase.java

@@ -0,0 +1,115 @@
+/**
+ * Copyright 2005-2026 Qlik
+ *<p>
+ * 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
+ *<p>
+ * Restlet is a registered trademark of QlikTech International AB.
+ */
+package org.restlet.engine.connector;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+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.Method;
+import org.restlet.data.Protocol;
+import org.restlet.engine.Engine;
+import org.restlet.representation.OutputRepresentation;
+
+/** Full response buffers reach the client while the producer pauses (issue #1632). */
+class FullBufferFlushTestCase {
+    /** The default output buffer size. */
+    private static final int BUFFER = 32 * 1024;
+
+    private Server server;
+    private Client client;
+
+    @BeforeEach
+    void setUp() {
+        Engine.clearThreadLocalVariables();
+        Engine.register(false);
+        Engine.getInstance().getRegisteredServers().add(new HttpServerHelper(null));
+        Engine.getInstance().getRegisteredClients().add(new HttpClientHelper(null));
+    }
+
+    @AfterEach
+    void tearDown() throws Exception {
+        if (client != null) {
+            client.stop();
+        }
+        if (server != null) {
+            server.stop();
+        }
+        Engine.clearThreadLocalVariables();
+    }
+
+    @ParameterizedTest
+    @CsvSource({"HTTP1_1, 1", "HTTP1_1, 2", "HTTP2, 1", "HTTP2, 2"})
+    void fullBuffersReachTheClientWhileTheProducerPauses(String mode, int buffers)
+            throws Exception {
+        int produced = buffers * BUFFER;
+        CountDownLatch read = new CountDownLatch(1);
+        AtomicBoolean heldBack = new AtomicBoolean();
+        Context serverContext = new Context();
+        serverContext.getParameters().add("http.transport.protocol", mode);
+        server =
+                new Server(
+                        serverContext,
+                        Protocol.HTTP,
+                        0,
+                        new Restlet() {
+                            @Override
+                            public void handle(Request request, Response response) {
+                                response.setEntity(
+                                        new OutputRepresentation(
+                                                MediaType.APPLICATION_OCTET_STREAM) {
+                                            @Override
+                                            public void write(OutputStream out) throws IOException {
+                                                out.write(new byte[produced]);
+                                                try {
+                                                    // Waits, without flushing, until the client
+                                                    // has read what was produced.
+                                                    heldBack.set(!read.await(5, TimeUnit.SECONDS));
+                                                } catch (InterruptedException e) {
+                                                    Thread.currentThread().interrupt();
+                                                }
+                                                out.write(new byte[16]);
+                                            }
+                                        });
+                            }
+                        });
+        server.start();
+        Context clientContext = new Context();
+        clientContext.getParameters().add("httpClientTransportMode", mode);
+        client = new Client(clientContext, Protocol.HTTP);
+        client.start();
+
+        Response response =
+                client.handle(
+                        new Request(Method.GET, "http://localhost:" + server.getActualPort()));
+        try (InputStream body = response.getEntity().getStream()) {
+            assertEquals(produced, body.readNBytes(produced).length);
+            read.countDown();
+            // The rest of the body follows the producer's wait, so its result is recorded.
+            assertEquals(16, body.readAllBytes().length);
+        }
+        assertFalse(heldBack.get(), "full buffers were held back until the producer resumed");
+    }
+}