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..4063000cf5 --- /dev/null +++ b/org.restlet/src/test/java/org/restlet/engine/connector/FullBufferFlushTestCase.java @@ -0,0 +1,115 @@ +/** + * 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.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"); + } +}