Return ByteFlow directly from factory method

Author: emil-bar

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

@@ -1,6 +1,10 @@
 package com.softwaremill.jox.flows;
 
-import static com.softwaremill.jox.structured.Scopes.unsupervised;
+import com.softwaremill.jox.*;
+import com.softwaremill.jox.flows.Flow.ByteFlow;
+import com.softwaremill.jox.structured.Fork;
+import com.softwaremill.jox.structured.Scopes;
+import com.softwaremill.jox.structured.ThrowingConsumer;
 
 import java.io.IOException;
 import java.io.InputStream;
@@ -11,11 +15,7 @@
 import java.nio.file.Path;
 import java.nio.file.StandardOpenOption;
 import java.time.Duration;
-import java.util.ArrayList;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
+import java.util.*;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.Flow.Publisher;
 import java.util.concurrent.Flow.Subscriber;
@@ -25,14 +25,7 @@
 import java.util.function.Function;
 import java.util.function.Supplier;
 
-import com.softwaremill.jox.Channel;
-import com.softwaremill.jox.ChannelClosed;
-import com.softwaremill.jox.ChannelDone;
-import com.softwaremill.jox.ChannelError;
-import com.softwaremill.jox.Source;
-import com.softwaremill.jox.structured.Fork;
-import com.softwaremill.jox.structured.Scopes;
-import com.softwaremill.jox.structured.ThrowingConsumer;
+import static com.softwaremill.jox.structured.Scopes.unsupervised;
 
 public final class Flows {
 
@@ -86,15 +79,15 @@ public static <T> Flow<T> fromValues(T... ts) {
      * Creates a ByteFlow from given {@link ByteChunk}s. Each ByteChunk is emitted in order.
      * Flow can be run multiple times.
      */
-    public static Flow.ByteFlow fromByteChunks(ByteChunk... chunks) {
+    public static ByteFlow fromByteChunks(ByteChunk... chunks) {
         return fromValues(chunks).toByteFlow();
     }
 
     /**
      * Creates a ByteFlow from given byte[]. Each byte[] is emitted in order.
      * Flow can be run multiple times.
      */
-    public static Flow.ByteFlow fromByteArrays(byte[]... byteArrays) {
+    public static ByteFlow fromByteArrays(byte[]... byteArrays) {
         return fromValues(byteArrays).toByteFlow();
     }
 
@@ -492,14 +485,14 @@ public static <T> Flow<T> interleaveAll(List<Flow<T>> flows, int segmentSize, bo
     }
 
     /**
-     * Converts a {@link java.io.InputStream} into a `Flow<ByteChunk>`.
+     * Converts a {@link java.io.InputStream} into {@link ByteFlow}.
      *
      * @param is
      *   an `InputStream` to read bytes from.
      * @param chunkSize
      *   maximum number of bytes to read from the underlying `InputStream` before emitting a new chunk.
      */
-    public static Flow<ByteChunk> fromInputStream(InputStream is, int chunkSize) {
+    public static ByteFlow fromInputStream(InputStream is, int chunkSize) {
         return usingEmit(emit -> {
             try (is) {
                 while (true) {
@@ -514,18 +507,18 @@ public static Flow<ByteChunk> fromInputStream(InputStream is, int chunkSize) {
                     }
                 }
             }
-        });
+        }).toByteFlow();
     }
 
     /**
-     * Creates a flow that emits {@link ByteChunk} read from a file.
+     * Creates a {@link ByteFlow} read from a file.
      *
      * @param path
      *   path the file to read from.
      * @param chunkSize
      *   maximum number of bytes to read from the file before emitting a new chunk.
      */
-    public static Flow<ByteChunk> fromFile(Path path, int chunkSize) {
+    public static ByteFlow fromFile(Path path, int chunkSize) {
         return usingEmit(emit -> {
             if (Files.isDirectory(path)) {
                 throw new IOException("Path %s is a directory".formatted(path));
@@ -553,6 +546,6 @@ public static Flow<ByteChunk> fromFile(Path path, int chunkSize) {
             } finally {
                 fileChannel.close();
             }
-        });
+        }).toByteFlow();
     }
 }