Return short reads from ByteFlow.runToInputStream

flows/src/main/java/com/softwaremill/jox/flows/Flow.java

@@ -2539,6 +2539,9 @@ public Flow<String> decodeStringUtf8() {
          * <p>Must be run within a concurrency scope, as under the hood the flow is run in the
          * background.
          *
+         * <p>Bulk reads block only until at least one byte is available; they may return fewer
+         * bytes than requested.
+         *
          * <p>Buffer capacity can be set via scoped value {@link Flow#CHANNEL_BUFFER_SIZE}. If not
          * specified in scope, {@link Channel#DEFAULT_BUFFER_SIZE} is used.
          */
@@ -2552,7 +2555,8 @@ public InputStream runToInputStream(Scope scope) throws InterruptedException {
                 private int availableBytes = 0;
                 private boolean isEndOfStream = false;
 
-                private boolean ensureDataAvailable() {
+                // does not block for the next chunk
+                private boolean advanceToBufferedByte() {
                     while (currentArrayIndex < currentArrays.size()) {
                         byte[] currentArray = currentArrays.get(currentArrayIndex);
                         if (currentByteIndex < currentArray.length) {
@@ -2561,6 +2565,13 @@ private boolean ensureDataAvailable() {
                         currentArrayIndex++;
                         currentByteIndex = 0;
                     }
+                    return false;
+                }
+
+                private boolean ensureDataAvailable() {
+                    if (advanceToBufferedByte()) {
+                        return true;
+                    }
 
                     if (!isEndOfStream) {
                         try {
@@ -2616,7 +2627,7 @@ public int read(byte[] b, int off, int len) {
                     int totalBytesRead = 0;
                     int remainingToRead = len;
 
-                    while (remainingToRead > 0 && ensureDataAvailable()) {
+                    while (remainingToRead > 0 && advanceToBufferedByte()) {
                         byte[] currentArray = currentArrays.get(currentArrayIndex);
                         int availableInCurrentArray = currentArray.length - currentByteIndex;
                         int bytesToRead = Math.min(remainingToRead, availableInCurrentArray);
@@ -2632,11 +2643,6 @@ public int read(byte[] b, int off, int len) {
                         totalBytesRead += bytesToRead;
                         remainingToRead -= bytesToRead;
                         availableBytes -= bytesToRead;
-
-                        if (currentByteIndex >= currentArray.length) {
-                            currentArrayIndex++;
-                            currentByteIndex = 0;
-                        }
                     }
 
                     return totalBytesRead;

flows/src/test/java/com/softwaremill/jox/flows/FlowIOTest.java

@@ -12,7 +12,9 @@
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
+import java.time.Duration;
 import java.util.List;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Function;
@@ -195,8 +197,63 @@ void handleEmptyChunksInStream() throws InterruptedException {
                     try (InputStream stream = source.runToInputStream(scope)) {
                         byte[] buffer = new byte[20];
                         int bytesRead = stream.read(buffer);
-                        assertEquals(13, bytesRead);
-                        assertEquals("Hello, World!", new String(buffer, 0, bytesRead));
+                        assertEquals(5, bytesRead);
+                        assertEquals("Hello", new String(buffer, 0, bytesRead));
+
+                        bytesRead = stream.read(buffer);
+                        assertEquals(2, bytesRead);
+                        assertEquals(", ", new String(buffer, 0, bytesRead));
+
+                        bytesRead = stream.read(buffer);
+                        assertEquals(6, bytesRead);
+                        assertEquals("World!", new String(buffer, 0, bytesRead));
+
+                        assertEquals(-1, stream.read(buffer));
+                    }
+                    return null;
+                });
+    }
+
+    @Test
+    void handleBulkReadWithoutWaitingForNextChunk() throws InterruptedException {
+        supervised(
+                scope -> {
+                    var release = new CountDownLatch(1);
+                    var source =
+                            Flows.<ByteChunk>usingEmit(
+                                            emit -> {
+                                                emit.apply(ByteChunk.fromArray("hello".getBytes()));
+                                                release.await();
+                                            })
+                                    .toByteFlow();
+                    try (InputStream stream = source.runToInputStream(scope)) {
+                        byte[] buffer = new byte[10];
+                        int bytesRead =
+                                assertTimeoutPreemptively(
+                                        Duration.ofSeconds(2), () -> stream.read(buffer));
+                        assertEquals(5, bytesRead);
+                        assertEquals("hello", new String(buffer, 0, bytesRead));
+                    } finally {
+                        // unblocks the producer so the scope can close
+                        release.countDown();
+                    }
+                    return null;
+                });
+    }
+
+    @Test
+    void handleBulkReadAcrossArraysOfSingleChunk() throws InterruptedException {
+        supervised(
+                scope -> {
+                    var chunk =
+                            ByteChunk.fromArray("ab".getBytes())
+                                    .concat(ByteChunk.fromArray("cd".getBytes()));
+                    var source = Flows.fromByteChunks(chunk);
+                    try (InputStream stream = source.runToInputStream(scope)) {
+                        byte[] buffer = new byte[10];
+                        int bytesRead = stream.read(buffer);
+                        assertEquals(4, bytesRead);
+                        assertEquals("abcd", new String(buffer, 0, bytesRead));
                     }
                     return null;
                 });