* This writer is a one-chunk-at-a-time pump driven entirely on the stream channel's event loop: *
* Lifecycle / cleanup. Because the pump completes asynchronously (after {@code writeHttp2}
@@ -123,6 +124,12 @@ default void onResume(Runnable resume) {
private ByteBuf pending;
private boolean done;
+ // flush() can synchronously fire channelWritabilityChanged. Prevent that callback from re-entering the
+ // pump, and remember permanently once the terminal frame has been emitted so later callbacks cannot write
+ // a second endStream frame before its write completes.
+ private boolean pumping;
+ private boolean terminalWritten;
+
// Transient handler that resumes the pump when the channel becomes writable again. Added lazily the
// first time the pump parks, removed by finish().
private WritabilityResumeHandler resumeHandler;
@@ -176,12 +183,14 @@ static void start(Http2StreamChannel channel, ChunkSource source) {
/**
* Produces and writes chunks until the channel goes unwritable (then parks for
- * {@code channelWritabilityChanged}) or the body is exhausted. Always runs on the event loop.
+ * {@code channelWritabilityChanged}), the source suspends, or the body is exhausted. Always runs on the
+ * event loop. Any written frames are flushed before the pump parks or completes.
*/
private void pump() {
- if (done) {
+ if (done || pumping || terminalWritten) {
return;
}
+ pumping = true;
try {
while (true) {
if (done) {
@@ -197,6 +206,12 @@ private void pump() {
// source signals more via the onResume callback. Any already-buffered `pending` chunk is
// retained (O(1)); we deliberately do not flush an early endStream.
suspended = true;
+ channel.flush();
+ if (!suspended) {
+ // A synchronous flush callback resumed the source while pump() was guarded against
+ // re-entry. Consume that resume here instead of parking indefinitely.
+ continue;
+ }
return;
}
@@ -208,6 +223,7 @@ private void pump() {
ByteBuf terminal = last != null ? last
// Empty body — preserve existing behaviour: a single empty DATA frame ends the stream.
: channel.alloc().buffer(0);
+ terminalWritten = true;
writeLastFrame(terminal);
channel.flush();
return;
@@ -219,9 +235,12 @@ private void pump() {
pending = next;
if (toWrite != null) {
writeFrame(toWrite, false);
- channel.flush();
if (!channel.isWritable()) {
+ channel.flush();
+ if (channel.isWritable()) {
+ continue;
+ }
// Flow-control window exhausted / high-water mark reached: stop producing and resume
// from channelWritabilityChanged. `pending` (one chunk) is retained until then.
ensureResumeHandler();
@@ -232,6 +251,8 @@ private void pump() {
}
} catch (Throwable t) {
finish(t);
+ } finally {
+ pumping = false;
}
}
@@ -308,7 +329,7 @@ private void finish(Throwable cause) {
private final class WritabilityResumeHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelWritabilityChanged(ChannelHandlerContext ctx) {
- if (!done && ctx.channel().isWritable()) {
+ if (!done && !terminalWritten && ctx.channel().isWritable()) {
pump();
}
ctx.fireChannelWritabilityChanged();
diff --git a/client/src/test/java/org/asynchttpclient/netty/request/body/Http2BodyWriterTest.java b/client/src/test/java/org/asynchttpclient/netty/request/body/Http2BodyWriterTest.java
new file mode 100644
index 000000000..670c78ad2
--- /dev/null
+++ b/client/src/test/java/org/asynchttpclient/netty/request/body/Http2BodyWriterTest.java
@@ -0,0 +1,347 @@
+/*
+ * Copyright (c) 2026 AsyncHttpClient Project. All rights reserved.
+ *
+ * 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 org.asynchttpclient.netty.request.body;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.UnpooledByteBufAllocator;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.channel.ChannelOutboundHandlerAdapter;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.ChannelPromise;
+import io.netty.channel.DefaultChannelPromise;
+import io.netty.channel.EventLoop;
+import io.netty.channel.WriteBufferWaterMark;
+import io.netty.channel.embedded.EmbeddedChannel;
+import io.netty.handler.codec.http2.DefaultHttp2DataFrame;
+import io.netty.handler.codec.http2.DefaultHttp2Headers;
+import io.netty.handler.codec.http2.DefaultHttp2HeadersFrame;
+import io.netty.handler.codec.http2.Http2DataFrame;
+import io.netty.handler.codec.http2.Http2FrameCodecBuilder;
+import io.netty.handler.codec.http2.Http2MultiplexHandler;
+import io.netty.handler.codec.http2.Http2StreamChannel;
+import io.netty.handler.codec.http2.Http2StreamChannelBootstrap;
+import io.netty.util.ReferenceCountUtil;
+import io.netty.util.concurrent.ImmediateEventExecutor;
+import org.junit.jupiter.api.Test;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+public class Http2BodyWriterTest {
+
+ @Test
+ public void multiChunkBodyBatchesFlushUntilTerminalFrame() {
+ Http2StreamChannel channel = mock(Http2StreamChannel.class);
+ EventLoop eventLoop = mock(EventLoop.class);
+ when(eventLoop.inEventLoop()).thenReturn(true);
+ when(channel.eventLoop()).thenReturn(eventLoop);
+ when(channel.alloc()).thenReturn(UnpooledByteBufAllocator.DEFAULT);
+ when(channel.isWritable()).thenReturn(true);
+ when(channel.closeFuture()).thenReturn(new DefaultChannelPromise(channel, ImmediateEventExecutor.INSTANCE));
+
+ AtomicInteger terminalFrames = new AtomicInteger();
+ when(channel.write(any())).thenAnswer(invocation -> {
+ Object msg = invocation.getArgument(0);
+ if (((DefaultHttp2DataFrame) msg).isEndStream()) {
+ terminalFrames.incrementAndGet();
+ }
+ ReferenceCountUtil.release(msg);
+ return succeededFuture(channel);
+ });
+
+ FixedChunkSource source = new FixedChunkSource(4);
+
+ Http2BodyWriter.start(channel, source);
+
+ assertEquals(1, source.closed);
+ assertEquals(1, terminalFrames.get());
+ verify(channel, times(4)).write(any(DefaultHttp2DataFrame.class));
+ verify(channel, times(1)).flush();
+ }
+
+ @Test
+ public void sourceSuspensionFlushesBeforeParking() {
+ Http2StreamChannel channel = mock(Http2StreamChannel.class);
+ EventLoop eventLoop = mock(EventLoop.class);
+ when(eventLoop.inEventLoop()).thenReturn(true);
+ when(channel.eventLoop()).thenReturn(eventLoop);
+ when(channel.alloc()).thenReturn(UnpooledByteBufAllocator.DEFAULT);
+ when(channel.isWritable()).thenReturn(true);
+ when(channel.closeFuture()).thenReturn(new DefaultChannelPromise(channel, ImmediateEventExecutor.INSTANCE));
+ AtomicInteger flushes = new AtomicInteger();
+ when(channel.write(any())).thenAnswer(invocation -> {
+ ReferenceCountUtil.release(invocation.getArgument(0));
+ return succeededFuture(channel);
+ });
+ when(channel.flush()).thenAnswer(invocation -> {
+ flushes.incrementAndGet();
+ return channel;
+ });
+
+ SuspendingChunkSource source = new SuspendingChunkSource(flushes);
+ Http2BodyWriter.start(channel, source);
+
+ assertEquals(0, source.closed);
+ assertEquals(0, source.flushesBeforeSuspend);
+ assertEquals(1, flushes.get());
+ verify(channel, times(1)).write(any(DefaultHttp2DataFrame.class));
+
+ source.finish();
+
+ assertEquals(1, source.closed);
+ verify(channel, times(2)).write(any(DefaultHttp2DataFrame.class));
+ assertEquals(2, flushes.get());
+ }
+
+ @Test
+ public void synchronousResumeDuringSuspensionFlushIsNotLost() {
+ Http2StreamChannel channel = mock(Http2StreamChannel.class);
+ EventLoop eventLoop = mock(EventLoop.class);
+ when(eventLoop.inEventLoop()).thenReturn(true);
+ when(channel.eventLoop()).thenReturn(eventLoop);
+ when(channel.alloc()).thenReturn(UnpooledByteBufAllocator.DEFAULT);
+ when(channel.isWritable()).thenReturn(true);
+ when(channel.closeFuture()).thenReturn(new DefaultChannelPromise(channel, ImmediateEventExecutor.INSTANCE));
+ when(channel.write(any())).thenAnswer(invocation -> {
+ ReferenceCountUtil.release(invocation.getArgument(0));
+ return succeededFuture(channel);
+ });
+ AtomicInteger flushes = new AtomicInteger();
+ SuspendingChunkSource source = new SuspendingChunkSource(flushes);
+ when(channel.flush()).thenAnswer(invocation -> {
+ if (flushes.incrementAndGet() == 1) {
+ source.finish();
+ }
+ return channel;
+ });
+
+ Http2BodyWriter.start(channel, source);
+
+ assertEquals(1, source.closed);
+ assertEquals(2, flushes.get());
+ verify(channel, times(2)).write(any(DefaultHttp2DataFrame.class));
+ }
+
+ @Test
+ public void realStreamBoundsUnflushedBytesByWaterMark() {
+ UnflushedBytesTracker tracker = new UnflushedBytesTracker();
+ EmbeddedChannel parent = new EmbeddedChannel(
+ Http2FrameCodecBuilder.forClient().build(),
+ new Http2MultiplexHandler(new ChannelInboundHandlerAdapter()));
+ Http2StreamChannel stream = new Http2StreamChannelBootstrap(parent)
+ .handler(tracker)
+ .open()
+ .syncUninterruptibly()
+ .getNow();
+ int chunkSize = 1024;
+ int highWaterMark = 8 * 1024;
+ stream.config().setWriteBufferWaterMark(new WriteBufferWaterMark(highWaterMark / 2, highWaterMark));
+ FixedChunkSource source = new FixedChunkSource(highWaterMark / chunkSize * 2, chunkSize);
+
+ try {
+ stream.writeAndFlush(new DefaultHttp2HeadersFrame(new DefaultHttp2Headers()
+ .method("POST")
+ .scheme("https")
+ .authority("localhost")
+ .path("/"))).syncUninterruptibly();
+
+ Http2BodyWriter.start(stream, source);
+ parent.runPendingTasks();
+
+ assertEquals(1, source.closed);
+ assertTrue(tracker.peakUnflushedBytes >= highWaterMark,
+ "the real stream must reach its configured high-water mark");
+ assertTrue(tracker.peakUnflushedBytes <= highWaterMark + chunkSize,
+ "unflushed DATA must stay within the high-water mark plus one chunk");
+ } finally {
+ stream.close().syncUninterruptibly();
+ parent.runPendingTasks();
+ parent.finishAndReleaseAll();
+ }
+ }
+
+ @Test
+ public void unwritableChannelResumesWithoutReentrantTerminalWrite() throws Exception {
+ Http2StreamChannel channel = mock(Http2StreamChannel.class);
+ EventLoop eventLoop = mock(EventLoop.class);
+ ChannelPipeline pipeline = mock(ChannelPipeline.class);
+ ChannelHandlerContext pipelineContext = mock(ChannelHandlerContext.class);
+ ChannelHandlerContext eventContext = mock(ChannelHandlerContext.class);
+ AtomicReference