From 5e70e89d7ee513f8f741187838617b604a2f1c5d Mon Sep 17 00:00:00 2001 From: Jerome Louvel <374450+jlouvel@users.noreply.github.com> Date: Tue, 3 Nov 2026 09:27:32 -0500 Subject: [PATCH 1/2] Flush each full Netty output buffer ChannelOutputStream sent a full buffer with channel.write() without flushing, so Netty kept it until a flush() or until more than the write buffer high-water mark was pending. A body whose producer paused after filling one or two buffers held back what it had produced. Full buffers are now written and flushed, so only a partially filled buffer waits for an explicit flush(). Fixes #1632. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Jerome Louvel <374450+jlouvel@users.noreply.github.com> --- changes.md | 3 + .../engine/netty/ChannelOutputStream.java | 17 +-- .../connector/FullBufferFlushTestCase.java | 112 ++++++++++++++++++ 3 files changed, 124 insertions(+), 8 deletions(-) create mode 100644 org.restlet/src/test/java/org/restlet/engine/connector/FullBufferFlushTestCase.java diff --git a/changes.md b/changes.md index cd2fc0ec14..a04160cda0 100644 --- a/changes.md +++ b/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. diff --git a/org.restlet/src/main/java/org/restlet/engine/netty/ChannelOutputStream.java b/org.restlet/src/main/java/org/restlet/engine/netty/ChannelOutputStream.java index b7e79884b3..63f8a6c7a2 100644 --- a/org.restlet/src/main/java/org/restlet/engine/netty/ChannelOutputStream.java +++ b/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. * *

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,7 +160,7 @@ 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; @@ -167,7 +168,7 @@ private void send(boolean flush) throws IOException { 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()); diff --git a/org.restlet/src/test/java/org/restlet/engine/connector/FullBufferFlushTestCase.java b/org.restlet/src/test/java/org/restlet/engine/connector/FullBufferFlushTestCase.java new file mode 100644 index 0000000000..4f1ee2c7a1 --- /dev/null +++ b/org.restlet/src/test/java/org/restlet/engine/connector/FullBufferFlushTestCase.java @@ -0,0 +1,112 @@ +/** + * 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.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.ValueSource; +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 { + /** Two full buffers of the default 32 KiB output buffer size. */ + private static final int PRODUCED = 64 * 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 + @ValueSource(strings = {"HTTP1_1", "HTTP2"}) + void fullBuffersReachTheClientWhileTheProducerPauses(String mode) throws Exception { + 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(); + assertFalse(heldBack.get(), "full buffers were held back until the producer resumed"); + assertEquals(16, body.readAllBytes().length); + } + } +} From 2e2ab446a28dcee3d49b57786f07aa7117b57930 Mon Sep 17 00:00:00 2001 From: Jerome Louvel <374450+jlouvel@users.noreply.github.com> Date: Tue, 3 Nov 2026 11:18:31 -0500 Subject: [PATCH 2/2] Test one and two full buffers, and check the wait after the body Co-Authored-By: Claude Opus 5.5 Signed-off-by: Jerome Louvel <374450+jlouvel@users.noreply.github.com> --- .../connector/FullBufferFlushTestCase.java | 19 +++++++++++-------- 1 file changed, 11 insertions(+), 8 deletions(-) diff --git a/org.restlet/src/test/java/org/restlet/engine/connector/FullBufferFlushTestCase.java b/org.restlet/src/test/java/org/restlet/engine/connector/FullBufferFlushTestCase.java index 4f1ee2c7a1..4063000cf5 100644 --- a/org.restlet/src/test/java/org/restlet/engine/connector/FullBufferFlushTestCase.java +++ b/org.restlet/src/test/java/org/restlet/engine/connector/FullBufferFlushTestCase.java @@ -20,7 +20,7 @@ import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.params.ParameterizedTest; -import org.junit.jupiter.params.provider.ValueSource; +import org.junit.jupiter.params.provider.CsvSource; import org.restlet.Client; import org.restlet.Context; import org.restlet.Request; @@ -35,8 +35,8 @@ /** Full response buffers reach the client while the producer pauses (issue #1632). */ class FullBufferFlushTestCase { - /** Two full buffers of the default 32 KiB output buffer size. */ - private static final int PRODUCED = 64 * 1024; + /** The default output buffer size. */ + private static final int BUFFER = 32 * 1024; private Server server; private Client client; @@ -61,8 +61,10 @@ void tearDown() throws Exception { } @ParameterizedTest - @ValueSource(strings = {"HTTP1_1", "HTTP2"}) - void fullBuffersReachTheClientWhileTheProducerPauses(String mode) throws Exception { + @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(); @@ -80,7 +82,7 @@ public void handle(Request request, Response response) { MediaType.APPLICATION_OCTET_STREAM) { @Override public void write(OutputStream out) throws IOException { - out.write(new byte[PRODUCED]); + out.write(new byte[produced]); try { // Waits, without flushing, until the client // has read what was produced. @@ -103,10 +105,11 @@ public void write(OutputStream out) throws IOException { client.handle( new Request(Method.GET, "http://localhost:" + server.getActualPort())); try (InputStream body = response.getEntity().getStream()) { - assertEquals(PRODUCED, body.readNBytes(PRODUCED).length); + assertEquals(produced, body.readNBytes(produced).length); read.countDown(); - assertFalse(heldBack.get(), "full buffers were held back until the producer resumed"); + // 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"); } }