HTTPCORE-799: Reset async entity producer state after interrupted writes

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();
     }
 
 }

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();
     }
 

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();
     }
 

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));
     }
 

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));
     }
 

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<ByteBuffer> 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));
+    }
+
 }

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<Header> trailers = dataStreamChannel.getTrailers();
+        Assertions.assertNotNull(trailers);
+        Assertions.assertEquals("digest", trailers.get(1).getName());
+        Assertions.assertEquals("e80b5017098950fc58aad83c8c14978e", trailers.get(1).getValue());
+    }
+
 }

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));
+    }
+
 }

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));
+    }
+
 }

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));
+    }
+
 }