DataChunkInputStream more then one close does not throw exception (helidon-io#1904)

Signed-off-by: David Kral <david.k.kral@oracle.com>

Author: Verdent

media/common/src/main/java/io/helidon/media/common/DataChunkInputStream.java

@@ -19,6 +19,7 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.nio.ByteBuffer;
+import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Flow;
@@ -97,7 +98,7 @@ private static void releaseChunk(DataChunk chunk, Throwable th) {
     @Override
     public void close() {
         // Assert: if current != next, next cannot ever be resolved with a chunk that needs releasing
-        current.whenComplete(DataChunkInputStream::releaseChunk);
+        Optional.ofNullable(current).ifPresent(it -> current.whenComplete(DataChunkInputStream::releaseChunk));
         current = null;
     }
 

media/common/src/test/java/io/helidon/media/common/DataChunkInputStreamTest.java

@@ -31,6 +31,7 @@
 import io.helidon.common.http.DataChunk;
 import io.helidon.common.reactive.BufferedEmittingPublisher;
 import io.helidon.common.reactive.Multi;
+import io.helidon.common.reactive.Single;
 
 import org.junit.jupiter.api.Test;
 
@@ -69,6 +70,13 @@ public void emptyChunk() throws IOException {
         assertThat(sb.toString(), is("foobar"));
     }
 
+    @Test
+    public void closeMoreTheOnce() throws IOException {
+        InputStream is = new DataChunkInputStream(Single.just(DataChunk.create("test".getBytes())));
+        is.close();
+        is.close();
+    }
+
     @Test
     public void differentThreads() throws Exception {
         List<String> test_data = List.of("test0", "test1", "test2", "test3");