Adds configurable buffering support to TcpClientConnection. Defaults to 4KB. New tests. See issue 9858. (helidon-io#9893)

Signed-off-by: Santiago Pericas-Geertsen <santiago.pericasgeertsen@oracle.com>

Author: spericas

common/buffers/src/main/java/io/helidon/common/buffers/DataWriter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2022 Oracle and/or its affiliates.
+ * Copyright (c) 2022, 2025 Oracle and/or its affiliates.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -21,7 +21,7 @@
  * Do not combine {@link #write(io.helidon.common.buffers.BufferData)} and {@link #writeNow(io.helidon.common.buffers.BufferData)}
  * to a single underlying transport, unless you can guarantee there will not be a race between these two methods.
  */
-public interface DataWriter {
+public interface DataWriter extends AutoCloseable {
     /**
      * Write buffers, may delay writing and may write on a different thread.
      * This method also may combine multiple calls into a single write to the underlying transport.
@@ -49,4 +49,19 @@ public interface DataWriter {
      * @param buffer buffer to write
      */
     void writeNow(BufferData buffer);
+
+    /**
+     * Flushes to the underlying transport any pending data that has been written using
+     * either {@link #write(BufferData)} or {@link #write(BufferData...)}.
+     */
+    default void flush() {
+    }
+
+    /**
+     * Closes this writer and frees any associated resources. Defaults to just a call
+     * to {@link #flush()}.
+     */
+    default void close() {
+        flush();
+    }
 }

http/http2/src/main/java/io/helidon/http/http2/Http2ConnectionWriter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2022, 2024 Oracle and/or its affiliates.
+ * Copyright (c) 2022, 2025 Oracle and/or its affiliates.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -56,7 +56,7 @@ public Http2ConnectionWriter(SocketContext ctx, DataWriter writer, List<Http2Fra
 
     @Override
     public void write(Http2FrameData frame) {
-           lockedWrite(frame);
+        lockedWrite(frame);
     }
 
     @Override
@@ -193,11 +193,11 @@ private void noLockWrite(Http2FrameData frame) {
         listener.frameHeader(ctx, streamId, headerData);
 
         if (frameHeader.length() == 0) {
-            writer.write(headerData);
+            writer.writeNow(headerData);
         } else {
             BufferData data = frame.data().copy();
             listener.frame(ctx, streamId, data);
-            writer.write(BufferData.create(headerData, data));
+            writer.writeNow(BufferData.create(headerData, data));
         }
     }
 

webclient/api/src/main/java/io/helidon/webclient/api/HttpClientConfigBlueprint.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2023, 2024 Oracle and/or its affiliates.
+ * Copyright (c) 2023, 2025 Oracle and/or its affiliates.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -266,4 +266,17 @@ default ClientRequestHeaders defaultRequestHeaders() {
     @Option.Configured
     @Option.DefaultInt(131072)
     int maxInMemoryEntity();
+
+    /**
+     * Buffer size used when writing data to the underlying socket on a client TCP
+     * connection. A value that is less or equal to 1 can be set to disable buffering
+     * at this level. Note that if writing data to the socket in small chunks, they
+     * may not be delivered to the network immediately due to Nagle's algorithm (i.e.,
+     * if TCP_NO_DELAY is turned off).
+     *
+     * @return number of bytes in write buffer
+     */
+    @Option.Configured
+    @Option.DefaultInt(4096)
+    int writeBufferSize();
 }

webclient/api/src/main/java/io/helidon/webclient/api/TcpClientConnection.java

@@ -146,8 +146,10 @@ Obtain target socket through proxy (if enabled), or connect to target socket
         } else {
             this.helidonSocket = PlainSocket.client(socket, channelId);
         }
+
         this.reader = new DataReader(helidonSocket);
-        this.writer = new DirectDatatWriter(helidonSocket);
+        int writeBufferSize = webClient.prototype().writeBufferSize();
+        this.writer = new BufferedDataWriter(helidonSocket, writeBufferSize);
 
         return this;
     }
@@ -310,33 +312,68 @@ private String certsToString(Certificate[] peerCertificates) {
         return String.join(", ", certs);
     }
 
-    private static class DirectDatatWriter implements DataWriter {
+    static class BufferedDataWriter implements DataWriter {
+        private final BufferData writerBuffer;
         private final HelidonSocket helidonSocket;
 
-        DirectDatatWriter(HelidonSocket helidonSocket) {
+        BufferedDataWriter(HelidonSocket helidonSocket, int bufferSize) {
             this.helidonSocket = helidonSocket;
+            this.writerBuffer = bufferSize > 0 ? BufferData.create(bufferSize) : null;
         }
 
         @Override
         public void write(BufferData... buffers) {
-            writeNow(buffers);
+            for (BufferData buffer : buffers) {
+                write(buffer);
+            }
         }
 
         @Override
         public void write(BufferData buffer) {
+            // no buffering, just write now
+            if (writerBuffer == null) {
+                writeNow(buffer);
+                return;
+            }
+            // have space in writerBuffer?
+            if (buffer.available() <= writerBuffer.capacity()) {
+                writerBuffer.write(buffer);
+                return;
+            }
+            // write and reset writerBuffer if we have data
+            if (writerBuffer.available() > 0) {
+                writeNow(writerBuffer);
+                writerBuffer.reset();
+                // have space now?
+                if (buffer.available() <= writerBuffer.capacity()) {
+                    writerBuffer.write(buffer);
+                    return;
+                }
+            }
+            // no luck, we need to write it now
             writeNow(buffer);
         }
 
         @Override
         public void writeNow(BufferData... buffers) {
+            flush();
             for (BufferData buffer : buffers) {
-                writeNow(buffer);
+                helidonSocket.write(buffer);
             }
         }
 
         @Override
         public void writeNow(BufferData buffer) {
+            flush();
             helidonSocket.write(buffer);
         }
+
+        @Override
+        public void flush() {
+            if (writerBuffer != null && writerBuffer.available() > 0) {
+                helidonSocket.write(writerBuffer);
+                writerBuffer.reset();
+            }
+        }
     }
 }

webclient/api/src/test/java/io/helidon/webclient/api/BufferedDataWriterTest.java

@@ -0,0 +1,179 @@
+/*
+ * Copyright (c) 2025 Oracle and/or its affiliates.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.helidon.webclient.api;
+
+import io.helidon.common.buffers.BufferData;
+import io.helidon.common.socket.HelidonSocket;
+import io.helidon.common.socket.PeerInfo;
+
+import org.junit.jupiter.api.Test;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+class BufferedDataWriterTest {
+
+    @Test
+    void testBufferFull() {
+        TestHelidonSocket socket = new TestHelidonSocket();
+        try (TcpClientConnection.BufferedDataWriter writer = new TcpClientConnection.BufferedDataWriter(socket, 5)) {
+            BufferData data = BufferData.create("0");
+            writer.write(data);
+            assertThat(socket.writeCounter(), is(0));
+
+            for (int i = 0; i < 4; i++) {
+                data.rewind();
+                writer.write(data);
+            }
+            assertThat(socket.writeCounter(), is(0));
+
+            data.rewind();
+            writer.write(data);
+            assertThat(socket.writeCounter(), is(5));
+
+            writer.flush();
+            assertThat(socket.writeCounter(), is(6));
+        }
+    }
+
+    @Test
+    void testNoFlush() {
+        TestHelidonSocket socket = new TestHelidonSocket();
+        try (TcpClientConnection.BufferedDataWriter writer = new TcpClientConnection.BufferedDataWriter(socket, 5)) {
+            BufferData data = BufferData.create("0");
+            for (int i = 0; i < 5; i++) {
+                data.rewind();
+                writer.write(data);
+            }
+            assertThat(socket.writeCounter(), is(0));
+        }
+        assertThat(socket.writeCounter(), is(5));
+    }
+
+    @Test
+    void testWritesAtBoundary() {
+        TestHelidonSocket socket = new TestHelidonSocket();
+        TcpClientConnection.BufferedDataWriter writer = new TcpClientConnection.BufferedDataWriter(socket, 5);
+        BufferData data = BufferData.create("0");
+        for (int i = 0; i < 50; i++) {
+            data.rewind();
+            writer.write(data);
+        }
+        writer.flush();
+        writer.close();
+        assertThat(socket.writeCounter(), is(50));
+    }
+
+    @Test
+    void testWritesOverBoundary() {
+        TestHelidonSocket socket = new TestHelidonSocket();
+        TcpClientConnection.BufferedDataWriter writer = new TcpClientConnection.BufferedDataWriter(socket, 5);
+        BufferData data = BufferData.create("0");
+        for (int i = 0; i < 52; i++) {
+            data.rewind();
+            writer.write(data);
+        }
+        writer.flush();
+        writer.close();
+        assertThat(socket.writeCounter(), is(52));
+    }
+
+    @Test
+    void testNoBuffering() {
+        TestHelidonSocket socket = new TestHelidonSocket();
+        TcpClientConnection.BufferedDataWriter writer = new TcpClientConnection.BufferedDataWriter(socket, -1);
+        BufferData data = BufferData.create("0");
+        for (int i = 0; i < 10; i++) {
+            data.rewind();
+            writer.write(data);
+            assertThat(socket.writeCounter(), is(i + 1));
+        }
+        writer.flush();
+        writer.close();
+    }
+
+    @Test
+    void testTooLargeToBuffer() {
+        TestHelidonSocket socket = new TestHelidonSocket();
+        TcpClientConnection.BufferedDataWriter writer = new TcpClientConnection.BufferedDataWriter(socket, 1);
+        BufferData data = BufferData.create("12345");
+        writer.write(data);     // can't buffer
+        assertThat(socket.writeCounter(), is(5));
+        writer.close();
+    }
+
+    static class TestHelidonSocket implements HelidonSocket {
+
+        private int writeCounter;
+
+        public int writeCounter() {
+            return writeCounter;
+        }
+
+        @Override
+        public void close() {
+        }
+
+        @Override
+        public void idle() {
+        }
+
+        @Override
+        public boolean isConnected() {
+            return false;
+        }
+
+        @Override
+        public int read(BufferData buffer) {
+            return 0;
+        }
+
+        @Override
+        public void write(BufferData buffer) {
+            writeCounter += buffer.available();
+        }
+
+        @Override
+        public PeerInfo remotePeer() {
+            return null;
+        }
+
+        @Override
+        public PeerInfo localPeer() {
+            return null;
+        }
+
+        @Override
+        public boolean isSecure() {
+            return false;
+        }
+
+        @Override
+        public String socketId() {
+            return null;
+        }
+
+        @Override
+        public String childSocketId() {
+            return null;
+        }
+
+        @Override
+        public byte[] get() {
+            return new byte[0];
+        }
+    }
+}

webclient/http1/src/main/java/io/helidon/webclient/http1/Http1CallEntityChain.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2023 Oracle and/or its affiliates.
+ * Copyright (c) 2023, 2025 Oracle and/or its affiliates.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -62,6 +62,7 @@ public WebClientServiceResponse doProceed(ClientConnection connection,
             writeBuffer.write(entity);
         }
         writer.write(writeBuffer);
+        writer.flush();
 
         return readResponse(serviceRequest, connection, reader);
     }