From 45f0404870e2423178ec6d50c36bf0a38bd6818b Mon Sep 17 00:00:00 2001 From: Matthias Kurz Date: Thu, 17 Sep 2026 14:05:45 +0200 Subject: [PATCH] HTTPCORE-799: Reset async entity producer state after interrupted writes Discard buffered but unwritten bytes and partial digest state when async entity producers are released, so repeating an interrupted production starts cleanly from the beginning. Add regression tests for file, path, string, repeatable binary, and digesting producers. --- .../AbstractBinAsyncEntityProducer.java | 1 + .../AbstractCharAsyncEntityProducer.java | 1 + .../nio/entity/DigestingEntityProducer.java | 1 + .../http/nio/entity/FileEntityProducer.java | 1 + .../http/nio/entity/PathEntityProducer.java | 1 + .../TestAbstractBinAsyncEntityProducer.java | 60 +++++++++++++++++++ .../entity/TestDigestingEntityProducer.java | 26 ++++++++ .../entity/TestFileAsyncEntityProducer.java | 18 ++++++ .../entity/TestPathAsyncEntityProducer.java | 19 ++++++ .../entity/TestStringAsyncEntityProducer.java | 19 ++++++ 10 files changed, 147 insertions(+) diff --git a/httpcore5/src/main/java/org/apache/hc/core5/http/nio/entity/AbstractBinAsyncEntityProducer.java b/httpcore5/src/main/java/org/apache/hc/core5/http/nio/entity/AbstractBinAsyncEntityProducer.java index b228d33977..60de28898f 100644 --- a/httpcore5/src/main/java/org/apache/hc/core5/http/nio/entity/AbstractBinAsyncEntityProducer.java +++ b/httpcore5/src/main/java/org/apache/hc/core5/http/nio/entity/AbstractBinAsyncEntityProducer.java @@ -223,6 +223,7 @@ public void endStream() throws IOException { @Override public void releaseResources() { state = State.ACTIVE; + byteBuffer.clear(); } } diff --git a/httpcore5/src/main/java/org/apache/hc/core5/http/nio/entity/AbstractCharAsyncEntityProducer.java b/httpcore5/src/main/java/org/apache/hc/core5/http/nio/entity/AbstractCharAsyncEntityProducer.java index 282ae9e602..63c1b4375f 100644 --- a/httpcore5/src/main/java/org/apache/hc/core5/http/nio/entity/AbstractCharAsyncEntityProducer.java +++ b/httpcore5/src/main/java/org/apache/hc/core5/http/nio/entity/AbstractCharAsyncEntityProducer.java @@ -245,6 +245,7 @@ public void endStream() throws IOException { @Override public void releaseResources() { state = State.ACTIVE; + bytebuf.clear(); charsetEncoder.reset(); } diff --git a/httpcore5/src/main/java/org/apache/hc/core5/http/nio/entity/DigestingEntityProducer.java b/httpcore5/src/main/java/org/apache/hc/core5/http/nio/entity/DigestingEntityProducer.java index 37e5b6fb71..172d71943e 100644 --- a/httpcore5/src/main/java/org/apache/hc/core5/http/nio/entity/DigestingEntityProducer.java +++ b/httpcore5/src/main/java/org/apache/hc/core5/http/nio/entity/DigestingEntityProducer.java @@ -155,6 +155,7 @@ public void failed(final Exception cause) { @Override public void releaseResources() { + digester.reset(); wrapped.releaseResources(); } diff --git a/httpcore5/src/main/java/org/apache/hc/core5/http/nio/entity/FileEntityProducer.java b/httpcore5/src/main/java/org/apache/hc/core5/http/nio/entity/FileEntityProducer.java index e56e90ff85..ac9400c944 100644 --- a/httpcore5/src/main/java/org/apache/hc/core5/http/nio/entity/FileEntityProducer.java +++ b/httpcore5/src/main/java/org/apache/hc/core5/http/nio/entity/FileEntityProducer.java @@ -155,6 +155,7 @@ public Exception getException() { @Override public void releaseResources() { eof = false; + byteBuffer.clear(); Closer.closeQuietly(accessFileRef.getAndSet(null)); } diff --git a/httpcore5/src/main/java/org/apache/hc/core5/http/nio/entity/PathEntityProducer.java b/httpcore5/src/main/java/org/apache/hc/core5/http/nio/entity/PathEntityProducer.java index 7a35d3ca73..28c1568437 100644 --- a/httpcore5/src/main/java/org/apache/hc/core5/http/nio/entity/PathEntityProducer.java +++ b/httpcore5/src/main/java/org/apache/hc/core5/http/nio/entity/PathEntityProducer.java @@ -161,6 +161,7 @@ public void produce(final DataStreamChannel dataStreamChannel) throws IOExceptio @Override public void releaseResources() { eof = false; + byteBuffer.clear(); Closer.closeQuietly(channelRef.getAndSet(null)); } diff --git a/httpcore5/src/test/java/org/apache/hc/core5/http/nio/entity/TestAbstractBinAsyncEntityProducer.java b/httpcore5/src/test/java/org/apache/hc/core5/http/nio/entity/TestAbstractBinAsyncEntityProducer.java index bb0c4d5fd2..ec2ff08f85 100644 --- a/httpcore5/src/test/java/org/apache/hc/core5/http/nio/entity/TestAbstractBinAsyncEntityProducer.java +++ b/httpcore5/src/test/java/org/apache/hc/core5/http/nio/entity/TestAbstractBinAsyncEntityProducer.java @@ -82,6 +82,48 @@ public void failed(final Exception cause) { } + static private class RepeatableByteAsyncEntityProducer extends AbstractBinAsyncEntityProducer { + + private final ByteBuffer content; + + public RepeatableByteAsyncEntityProducer( + final int fragmentSizeHint, + final ContentType contentType, + final byte[] content) { + super(fragmentSizeHint, contentType); + this.content = ByteBuffer.wrap(content); + } + + @Override + public boolean isRepeatable() { + return true; + } + + @Override + protected int availableData() { + return Integer.MAX_VALUE; + } + + @Override + protected void produceData(final StreamChannel channel) throws IOException { + channel.write(content); + if (!content.hasRemaining()) { + channel.endStream(); + } + } + + @Override + public void failed(final Exception cause) { + } + + @Override + public void releaseResources() { + content.clear(); + super.releaseResources(); + } + + } + @Test void testProduceDataNoBuffering() throws Exception { @@ -184,4 +226,22 @@ void testProduceDataWithBuffering2() throws Exception { } + @Test + void testProduceDataRepeatableAfterPartialWrite() throws Exception { + final AsyncEntityProducer producer = new RepeatableByteAsyncEntityProducer( + 6, ContentType.TEXT_PLAIN, "abcdef".getBytes(StandardCharsets.US_ASCII)); + + final WritableByteChannelMock partialByteChannel = new WritableByteChannelMock(1024, 3); + producer.produce(new BasicDataStreamChannel(partialByteChannel)); + Assertions.assertEquals("abc", partialByteChannel.dump(StandardCharsets.US_ASCII)); + producer.releaseResources(); + + final WritableByteChannelMock byteChannel = new WritableByteChannelMock(1024); + final DataStreamChannel streamChannel = new BasicDataStreamChannel(byteChannel); + producer.produce(streamChannel); + + Assertions.assertFalse(byteChannel.isOpen()); + Assertions.assertEquals("abcdef", byteChannel.dump(StandardCharsets.US_ASCII)); + } + } diff --git a/httpcore5/src/test/java/org/apache/hc/core5/http/nio/entity/TestDigestingEntityProducer.java b/httpcore5/src/test/java/org/apache/hc/core5/http/nio/entity/TestDigestingEntityProducer.java index 24b33931e4..2d0ec05969 100644 --- a/httpcore5/src/test/java/org/apache/hc/core5/http/nio/entity/TestDigestingEntityProducer.java +++ b/httpcore5/src/test/java/org/apache/hc/core5/http/nio/entity/TestDigestingEntityProducer.java @@ -34,6 +34,7 @@ import org.apache.hc.core5.http.Header; import org.apache.hc.core5.http.WritableByteChannelMock; import org.apache.hc.core5.http.nio.BasicDataStreamChannel; +import org.apache.hc.core5.util.TextUtils; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; @@ -62,4 +63,29 @@ void testProduceData() throws Exception { Assertions.assertEquals("827ccb0eea8a706c4c34a16891f84e7b", trailers.get(1).getValue()); } + @Test + void testProduceDataRepeatableAfterPartialWrite() throws Exception { + final DigestingEntityProducer producer = new DigestingEntityProducer("MD5", + new StringAsyncEntityProducer("abcdef", 6, 6, ContentType.TEXT_PLAIN)); + + final WritableByteChannelMock partialByteChannel = new WritableByteChannelMock(1024, 3); + producer.produce(new BasicDataStreamChannel(partialByteChannel)); + Assertions.assertEquals("abc", partialByteChannel.dump(StandardCharsets.US_ASCII)); + producer.releaseResources(); + + final WritableByteChannelMock byteChannel = new WritableByteChannelMock(1024); + final BasicDataStreamChannel dataStreamChannel = new BasicDataStreamChannel(byteChannel); + while (byteChannel.isOpen()) { + producer.produce(dataStreamChannel); + } + + Assertions.assertEquals("abcdef", byteChannel.dump(StandardCharsets.US_ASCII)); + Assertions.assertEquals("e80b5017098950fc58aad83c8c14978e", TextUtils.toHexString(producer.getDigest())); + + final List
trailers = dataStreamChannel.getTrailers(); + Assertions.assertNotNull(trailers); + Assertions.assertEquals("digest", trailers.get(1).getName()); + Assertions.assertEquals("e80b5017098950fc58aad83c8c14978e", trailers.get(1).getValue()); + } + } diff --git a/httpcore5/src/test/java/org/apache/hc/core5/http/nio/entity/TestFileAsyncEntityProducer.java b/httpcore5/src/test/java/org/apache/hc/core5/http/nio/entity/TestFileAsyncEntityProducer.java index 77ff3e94de..442032ad21 100644 --- a/httpcore5/src/test/java/org/apache/hc/core5/http/nio/entity/TestFileAsyncEntityProducer.java +++ b/httpcore5/src/test/java/org/apache/hc/core5/http/nio/entity/TestFileAsyncEntityProducer.java @@ -104,4 +104,22 @@ void testTextContentRepeatable() throws Exception { } } + @Test + void testTextContentRepeatableAfterPartialWrite() throws Exception { + final AsyncEntityProducer producer = new FileEntityProducer(tempFile, ContentType.TEXT_PLAIN); + + final WritableByteChannelMock partialByteChannel = new WritableByteChannelMock(1024, 3); + producer.produce(new BasicDataStreamChannel(partialByteChannel)); + Assertions.assertEquals("abc", partialByteChannel.dump(StandardCharsets.US_ASCII)); + producer.releaseResources(); + + final WritableByteChannelMock byteChannel = new WritableByteChannelMock(1024); + final DataStreamChannel streamChannel = new BasicDataStreamChannel(byteChannel); + producer.produce(streamChannel); + producer.produce(streamChannel); + + Assertions.assertFalse(byteChannel.isOpen()); + Assertions.assertEquals("abcdef", byteChannel.dump(StandardCharsets.US_ASCII)); + } + } diff --git a/httpcore5/src/test/java/org/apache/hc/core5/http/nio/entity/TestPathAsyncEntityProducer.java b/httpcore5/src/test/java/org/apache/hc/core5/http/nio/entity/TestPathAsyncEntityProducer.java index 211ddd7499..975603f582 100644 --- a/httpcore5/src/test/java/org/apache/hc/core5/http/nio/entity/TestPathAsyncEntityProducer.java +++ b/httpcore5/src/test/java/org/apache/hc/core5/http/nio/entity/TestPathAsyncEntityProducer.java @@ -109,4 +109,23 @@ void testTextContentRepeatable() throws Exception { } } + @Test + void testTextContentRepeatableAfterPartialWrite() throws Exception { + final Path tempPath = tempFile.toPath(); + final AsyncEntityProducer producer = new PathEntityProducer(tempPath, ContentType.TEXT_PLAIN, StandardOpenOption.READ); + + final WritableByteChannelMock partialByteChannel = new WritableByteChannelMock(1024, 3); + producer.produce(new BasicDataStreamChannel(partialByteChannel)); + Assertions.assertEquals("abc", partialByteChannel.dump(StandardCharsets.US_ASCII)); + producer.releaseResources(); + + final WritableByteChannelMock byteChannel = new WritableByteChannelMock(1024); + final DataStreamChannel streamChannel = new BasicDataStreamChannel(byteChannel); + producer.produce(streamChannel); + producer.produce(streamChannel); + + Assertions.assertFalse(byteChannel.isOpen()); + Assertions.assertEquals("abcdef", byteChannel.dump(StandardCharsets.US_ASCII)); + } + } diff --git a/httpcore5/src/test/java/org/apache/hc/core5/http/nio/entity/TestStringAsyncEntityProducer.java b/httpcore5/src/test/java/org/apache/hc/core5/http/nio/entity/TestStringAsyncEntityProducer.java index 20df457384..abc3c1cfc3 100644 --- a/httpcore5/src/test/java/org/apache/hc/core5/http/nio/entity/TestStringAsyncEntityProducer.java +++ b/httpcore5/src/test/java/org/apache/hc/core5/http/nio/entity/TestStringAsyncEntityProducer.java @@ -80,4 +80,23 @@ void testTextContentRepeatable() throws Exception { } } + @Test + void testTextContentRepeatableAfterPartialWrite() throws Exception { + final AsyncEntityProducer producer = new StringAsyncEntityProducer( + "abcdef", 6, 6, ContentType.TEXT_PLAIN); + + final WritableByteChannelMock partialByteChannel = new WritableByteChannelMock(1024, 3); + producer.produce(new BasicDataStreamChannel(partialByteChannel)); + Assertions.assertEquals("abc", partialByteChannel.dump(StandardCharsets.US_ASCII)); + producer.releaseResources(); + + final WritableByteChannelMock byteChannel = new WritableByteChannelMock(1024); + final DataStreamChannel streamChannel = new BasicDataStreamChannel(byteChannel); + producer.produce(streamChannel); + producer.produce(streamChannel); + + Assertions.assertFalse(byteChannel.isOpen()); + Assertions.assertEquals("abcdef", byteChannel.dump(StandardCharsets.US_ASCII)); + } + }