Repository: softwaremill/jox (java)

Issue:
Add remaing Flow methods
Add folowing flow methods
Sample
Debounce
DebounceBy
Collect
Scan
Zip
zipAll
zipWithIndex

Pull request description:
Add remaining flow methods
Closes #84 

Patch:
diff --git a/flows/src/main/java/com/softwaremill/jox/flows/Flow.java b/flows/src/main/java/com/softwaremill/jox/flows/Flow.java
index 016dae7..591445c 100644
--- a/flows/src/main/java/com/softwaremill/jox/flows/Flow.java
+++ b/flows/src/main/java/com/softwaremill/jox/flows/Flow.java
@@ -52,6 +52,8 @@ import com.softwaremill.jox.Source;
 import com.softwaremill.jox.structured.CancellableFork;
 import com.softwaremill.jox.structured.Fork;
 import com.softwaremill.jox.structured.Scope;
+import com.softwaremill.jox.structured.Scopes;
+import com.softwaremill.jox.structured.ThrowingRunnable;
 import com.softwaremill.jox.structured.UnsupervisedScope;
 
 /**
@@ -439,6 +441,290 @@ public class Flow<T> {
         });
     }
 
+    /**
+     * Emits only every nth element emitted by this flow.
+     *
+     * @param n
+     *   The interval between two emitted elements.
+     */
+    public Flow<T> sample(int n) {
+        return Flows.usingEmit(emit -> {
+            AtomicInteger sampleCounter = new AtomicInteger(0);
+            last.run(t -> {
+                int counter = sampleCounter.incrementAndGet();
+                if (n != 0 && counter % n == 0) {
+                    emit.apply(t);
+                }
+            });
+        });
+    }
+
+    /** Remove subsequent, repeating elements
+     */
+    public Flow<T> debounce() {
+        return debounceBy(Function.identity());
+    }
+
+    /**
+     * Remove subsequent, repeating elements matching 'f'
+     *
+     * @param f The function used to compare the previous and current elements
+     */
+    public <U> Flow<T> debounceBy(Function<T, U> f) {
+        return Flows.usingEmit(emit -> {
+            AtomicReference<Optional<U>> previousElement = new AtomicReference<>(Optional.empty());
+            last.run(t -> {
+                U currentElement = f.apply(t);
+                if (!previousElement.get().equals(Optional.ofNullable(currentElement))) {
+                    emit.apply(t);
+                }
+                previousElement.set(Optional.ofNullable(currentElement));
+            });
+        });
+    }
+
+    /**
+     * Applies the given mapping function `f` to each element emitted by this flow, for which the function returns a non-empty Optional, and emits the result.
+     * If `f` returns an empty Optional at an element, the element will be skipped.
+     *
+     * @param f
+     *   The mapping function.
+     */
+    public <U> Flow<U> collect(Function<T, Optional<U>> f) {
+        return Flows.usingEmit(emit ->
+                last.run(t -> {
+                    Optional<U> result = f.apply(t);
+                    if (result.isPresent()) {
+                        emit.apply(result.get());
+                    }
+                })
+        );
+    }
+
+    /**
+     * Transforms the elements of the flow by applying an accumulation function to each element, producing a new value at each step. The
+     * resulting flow contains the accumulated values at each point in the original flow.
+     *
+     * @param initial The initial value to start the accumulation.
+     * @param f The accumulation function that is applied to each element of the flow.
+     * @return A new Flow containing the accumulated values.
+     */
+    public <V> Flow<V> scan(V initial, BiFunction<V, T, V> f) {
+        return Flows.usingEmit(emit -> {
+            emit.apply(initial);
+            AtomicReference<V> accumulator = new AtomicReference<>(initial);
+            last.run(t -> {
+                emit.apply(accumulator.updateAndGet(acc -> f.apply(acc, t)));
+            });
+        });
+    }
+
+    /** Combines elements from this and the other flow into Map.Entry. Completion of either flow completes the returned flow as well. The flows
+     * are run concurrently.
+     * <p>
+     * Method uses channels to emit elements. The size of channel buffer is determined by the scoped value {@link Channel#BUFFER_SIZE} or {@link Channel#DEFAULT_BUFFER_SIZE} is used.
+     * 
+     * @see
+     *   Flow#zipAll 
+     */
+    public <U> Flow<Map.Entry<T, U>> zip(Flow<U> other) {
+        return Flows.usingEmit(emit -> {
+            Scopes.unsupervised(scope -> {
+                Source<T> s1 = this.runToChannel(scope);
+                Source<U> s2 = other.runToChannel(scope);
+
+                while (true) {
+                    Object result1 = s1.receiveOrClosed();
+                    if (result1 instanceof ChannelDone) {
+                        return null;
+                    } else if (result1 instanceof ChannelError error) {
+                        throw error.toException();
+                    } else {
+                        // noinspection unchecked
+                        T t = (T) result1;
+                        Object result2 = s2.receiveOrClosed();
+                        if (result2 instanceof ChannelDone) {
+                            return null;
+                        } else if (result2 instanceof ChannelError error) {
+                            throw error.toException();
+                        } else {
+                            // noinspection unchecked
+                            U u = (U) result2;
+                            emit.apply(Map.entry(t, u));
+                        }
+                    }
+                }
+            });
+        });
+    }
+
+    /**
+     * Combines elements from this and the other flow into tuples, handling early completion of either flow with defaults. The flows are run
+     * concurrently.
+     * <p>
+     * Method uses channels to emit elements. The size of channel buffer is determined by the scoped value {@link Channel#BUFFER_SIZE} or {@link Channel#DEFAULT_BUFFER_SIZE} is used.
+     * 
+     * @param other
+     *   A flow of elements to be combined with.
+     * @param thisDefault
+     *   A default element to be used in the result tuple when the other flow is longer.
+     * @param otherDefault
+     *   A default element to be used in the result tuple when the current flow is longer.
+     */
+    public <U> Flow<Map.Entry<T, U>> zipAll(Flow<U> other, T thisDefault, U otherDefault) {
+        return Flows.usingEmit(emit -> {
+            Scopes.unsupervised(scope -> {
+                Source<T> s1 = this.runToChannel(scope);
+                Source<U> s2 = other.runToChannel(scope);
+
+                boolean continueLoop = true;
+                while (continueLoop) {
+                    Object received1 = s1.receiveOrClosed();
+                    if (received1 instanceof ChannelDone) {
+                        Object received2 = s2.receiveOrClosed();
+                        if (received2 instanceof ChannelDone) {
+                            continueLoop = false;
+                        } else if (received2 instanceof ChannelError e) {
+                            throw e.toException();
+                        } else {
+                            //noinspection unchecked
+                            emit.apply(Map.entry(thisDefault, (U) received2));
+                        }
+                    } else if (received1 instanceof ChannelError e) {
+                        throw e.toException();
+                    } else {
+                        Object received2 = s2.receiveOrClosed();
+                        if (received2 instanceof ChannelDone) {
+                            //noinspection unchecked
+                            emit.apply(Map.entry((T) received1, otherDefault));
+                        } else if (received2 instanceof ChannelError e) {
+                            throw e.toException();
+                        } else {
+                            //noinspection unchecked
+                            emit.apply(Map.entry((T) received1, (U) received2));
+                        }
+                    }
+                }
+                return null;
+            });
+        });
+    }
+
+    /**
+     * Combines each element from this and the index of the element (starting at 0).
+     */
+    public Flow<Map.Entry<T, Long>> zipWithIndex() {
+        return Flows.usingEmit(emit -> {
+            AtomicLong index = new AtomicLong(0L);
+            last.run(t -> {
+                Map.Entry<T, Long> zipped = Map.entry(t, index.getAndIncrement());
+                emit.apply(zipped);
+            });
+        });
+    }
+
+    /** Given that this flow emits other flows, flattens the nested flows into a single flow. The resulting flow emits elements from the
+     * nested flows in the order they are emitted.
+     * <p>
+     * The nested flows are run in sequence, that is, the next nested flow is started only after the previous one completes.
+     *
+     * @param args
+     *   This param should *NOT* be passed. It's only used to verify that this flow contains other flows.
+     * @throws IllegalArgumentException
+     *   when flow does not contain nested flows, or when `args` are not empty
+     */
+    @SafeVarargs
+    public final T flatten(T... args) {
+        if (!Flow.class.equals(getTClass(args))) {
+            throw new IllegalArgumentException("requirement failed: flatten can be called on Flow containing Flows");
+        }
+        //noinspection unchecked,rawtypes
+        return (T) this.flatMap(t -> (Flow) t);
+    }
+
+    /** Pipes the elements of child flows into the returned flow.
+     * <p>
+     * If this flow or any of the child flows emit an error, the pulling stops and the output flow propagates the error.
+     * <p>
+     * Up to `parallelism` child flows are run concurrently in the background. When the limit is reached, until a child flow completes, no
+     * more child flows are run.
+     * <p>
+     * The size of the buffers for the elements emitted by the child flows is determined by the {@link Channel#BUFFER_SIZE} that is in scope, or default {@link Channel#DEFAULT_BUFFER_SIZE} is used.
+     *
+     * @param parallelism
+     *   An upper bound on the number of child flows that run in parallel.
+     * @param args
+     *   This param should *NOT* be passed. It's only used to verify that this flow contains other flows.
+     * @throws IllegalArgumentException
+     *   when flow does not contain nested flows, or when `args` are not empty
+     */
+    @SuppressWarnings("unchecked")
+    @SafeVarargs
+    public final <U> T flattenPar(int parallelism, T... args) {
+        if (!Flow.class.equals(getTClass(args))) {
+            throw new IllegalArgumentException("requirement failed: flattenPar can be called on Flow containing Flows");
+        }
+        return (T) Flows.usingEmit(emit -> {
+            class Nested {
+                final Flow<U> child;
+                Nested(Flow<U> child) {
+                    this.child = child;
+                }
+            }
+            final class ChildDone {}
+
+            unsupervised(scope -> {
+                Channel<U> childOutputChannel = Channel.withScopedBufferSize();
+                Channel<ChildDone> childDoneChannel = Channel.withScopedBufferSize();
+
+                // When an error occurs in the parent, propagating it also to `childOutputChannel`, from which we always
+                // `select` in the main loop. That way, even if max parallelism is reached, errors in the parent will
+                // be discovered without delay.
+                //noinspection unchecked
+                Source<Nested> parentChannel = map(t -> new Nested((Flow<U>) t))
+                        .onError(childOutputChannel::error)
+                        .runToChannel(scope);
+
+                int runningChannelCount = 1; // parent is running
+                boolean parentDone = false;
+
+                while (runningChannelCount > 0) {
+                    assert runningChannelCount <= parallelism + 1;
+
+                    Object result;
+                    if (runningChannelCount == parallelism + 1 || parentDone) {
+                        result = selectOrClosed(childOutputChannel.receiveClause(), childDoneChannel.receiveClause());
+                    } else {
+                        result = selectOrClosed(childOutputChannel.receiveClause(), childDoneChannel.receiveClause(), parentChannel.receiveClause());
+                    }
+
+                    // Only `parentChannel` might be done, child completion is signalled via `childDoneChannel`.
+                    if (result instanceof ChannelDone) {
+                        parentDone = parentChannel.isClosedForReceive();
+                        assert parentDone;
+                        runningChannelCount--;
+                    } else if (result instanceof ChannelError e) {
+                        throw e.toException();
+                    } else if (ChildDone.class.isInstance(result)) {
+                        runningChannelCount--;
+                    } else if (Nested.class.isInstance(result)) {
+                        //noinspection unchecked
+                        Nested t = (Nested) result;
+                        scope.forkUnsupervised(() -> {
+                            t.child.onDone(() -> childDoneChannel.send(new ChildDone()))
+                                    .runPipeToSink(childOutputChannel, false);
+                            return null;
+                        });
+                        runningChannelCount++;
+                    } else if (result != null) {
+                        emit.apply(result);
+                    }
+                }
+                return null;
+            });
+        });
+    }
+
     /**
      * Applies the given `mappingFunction` to each element emitted by this flow, in sequence.
      * The given {@link Consumer<FlowEmit>} can be used to emit an arbitrary number of elements.
@@ -715,7 +1001,7 @@ public class Flow<T> {
     /**
      * Runs `f` after the flow completes successfully, that is when all elements are emitted.
      */
-    public Flow<T> onDone(Runnable f) {
+    public Flow<T> onDone(ThrowingRunnable f) {
         return usingEmit(emit -> {
             last.run(emit);
             f.run();
@@ -1221,19 +1507,20 @@ public class Flow<T> {
     // endregion
 
     // region ByteFlow
-
     public interface ByteChunkMapper<T> extends Function<T, ByteChunk> {}
     public interface ByteArrayMapper<T> extends Function<T, byte[]> {}
 
     /**
      * Converts a flow of `byte[]` or {@link ByteChunk} into a dedicated Flow type {@link ByteFlow}.
      *
-     * @throws IllegalArgumentException if the flow does not contain `byte[]` or {@link ByteChunk} elements.
+     * @param args
+     *   This param should *NOT* be passed. It's only used to verify that this flow contains byte[] or {@link ByteChunk}.
+     * @throws IllegalArgumentException
+     *   if the flow does not contain `byte[]` or {@link ByteChunk} elements.
      */
     @SafeVarargs
     public final ByteFlow toByteFlow(T... args) {
-        //noinspection unchecked
-        return new ByteFlow(last, (Class<T>) args.getClass().getComponentType());
+        return new ByteFlow(last, getTClass(args));
     }
 
     /**
@@ -1456,6 +1743,14 @@ public class Flow<T> {
         });
     }
 
+    @SuppressWarnings("unchecked")
+    private static <T> Class<T> getTClass(T[] args) {
+        if (args.length > 0) {
+            throw new IllegalArgumentException("Please do not pass any arguments for this method. Java will detect the type automatically.");
+        }
+        return (Class<T>) args.getClass().getComponentType();
+    }
+
     private static class BreakException extends RuntimeException {
     }
 }
diff --git a/flows/src/main/java/com/softwaremill/jox/flows/Flows.java b/flows/src/main/java/com/softwaremill/jox/flows/Flows.java
index 4026b83..3c299b8 100644
--- a/flows/src/main/java/com/softwaremill/jox/flows/Flows.java
+++ b/flows/src/main/java/com/softwaremill/jox/flows/Flows.java
@@ -12,7 +12,6 @@ import java.nio.file.Path;
 import java.nio.file.StandardOpenOption;
 import java.time.Duration;
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
@@ -60,29 +59,48 @@ public final class Flows {
 
     /**
      * Creates a flow from the given `iterable`. Each element of the iterable is emitted in order.
+     * The flow can be run multiple times.
      */
     public static <T> Flow<T> fromIterable(Iterable<T> iterable) {
-        return fromIterator(iterable.iterator());
+        return usingEmit(emit -> {
+            for (T t : iterable) {
+                emit.apply(t);
+            }
+        });
     }
 
     /**
      * Creates a flow from the given values. Each value is emitted in order.
+     * Flow can be run multiple times.
      */
     @SafeVarargs
     public static <T> Flow<T> fromValues(T... ts) {
-        return fromIterator(Arrays.asList(ts).iterator());
+        return usingEmit(emit -> {
+            for (T t : ts) {
+                emit.apply(t);
+            }
+        });
     }
 
+    /**
+     * 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) {
-        return fromIterator(Arrays.asList(chunks).iterator()).toByteFlow();
+        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) {
         return fromValues(byteArrays).toByteFlow();
     }
 
     /**
      * Creates a flow from the given (lazily evaluated) `iterator`. Each element of the iterator is emitted in order.
+     * The flow can be run only once, as the iterator is consumed. If you need to run the flow multiple times, use {@link #fromIterator(Supplier)}.
      */
     public static <T> Flow<T> fromIterator(Iterator<T> it) {
         return usingEmit(emit -> {
@@ -92,6 +110,19 @@ public final class Flows {
         });
     }
 
+    /**
+     * Creates a flow from the given (lazily evaluated) `iterator`. Each element of the iterator is emitted in order.
+     * The flow can be run multiple times, as the `iteratorSupplier` is called each time the flow is run.
+     */
+    public static <T> Flow<T> fromIterator(Supplier<Iterator<T>> iteratorSupplier) {
+        return usingEmit(emit -> {
+            var it = iteratorSupplier.get();
+            while (it.hasNext()) {
+                emit.apply(it.next());
+            }
+        });
+    }
+
     /**
      * Creates a flow from the given fork. The flow will emit up to one element, or complete by throwing an exception if the fork fails.
      */
diff --git a/structured/src/main/java/com/softwaremill/jox/structured/Scope.java b/structured/src/main/java/com/softwaremill/jox/structured/Scope.java
index 526b9ab..b4259ff 100644
--- a/structured/src/main/java/com/softwaremill/jox/structured/Scope.java
+++ b/structured/src/main/java/com/softwaremill/jox/structured/Scope.java
@@ -3,7 +3,6 @@ package com.softwaremill.jox.structured;
 import java.util.concurrent.Callable;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.StructuredTaskScope;
-import java.util.concurrent.atomic.AtomicReference;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 


Test patch:
diff --git a/flows/src/test/java/com/softwaremill/jox/flows/FlowFlattenTest.java b/flows/src/test/java/com/softwaremill/jox/flows/FlowFlattenTest.java
new file mode 100644
index 0000000..546b746
--- /dev/null
+++ b/flows/src/test/java/com/softwaremill/jox/flows/FlowFlattenTest.java
@@ -0,0 +1,129 @@
+package com.softwaremill.jox.flows;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.contains;
+import static org.hamcrest.Matchers.containsInAnyOrder;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.hamcrest.Matchers;
+import org.junit.jupiter.api.Test;
+
+public class FlowFlattenTest {
+
+    @Test
+    void flattenTest() throws Exception {
+        // given
+        Flow<Flow<Integer>> flow = Flows.fromValues(Flows.fromValues(1, 2), Flows.fromValues(5, 9));
+
+        // when
+        List<Integer> integers = flow.flatten().runToList();
+
+        // then
+        assertEquals(List.of(1, 2, 5, 9), integers);
+    }
+
+    @Test
+    void flatten_shouldThrowWhenCalledOnFlowNotContainingFlows() {
+        // when & then
+        IllegalArgumentException exception = assertThrows(IllegalArgumentException.class, () -> Flows.fromValues(3, 3).flatten());
+        assertEquals("requirement failed: flatten can be called on Flow containing Flows", exception.getMessage());
+    }
+
+    @Test
+    void flattenPar_shouldThrowWhenCalledOnFlowNotContainingFlows() {
+        // when & then
+        IllegalArgumentException exception = assertThrows(IllegalArgumentException.class,
+                () -> Flows.fromValues(3, 3).flattenPar(1));
+        assertEquals("requirement failed: flattenPar can be called on Flow containing Flows", exception.getMessage());
+    }
+
+    @Test
+    void shouldPipeAllElementsOfTheChildFlowsIntoTheOutputFlow() throws Exception {
+        // given
+        var flow = Flows.fromValues(
+                Flows.fromValues(10),
+                Flows.fromValues(20, 30),
+                Flows.fromValues(40, 50, 60)
+        );
+
+        // when & then
+        List<Integer> actual = flow.flattenPar(10).runToList();
+        assertThat(actual, containsInAnyOrder(10, 20, 30, 40, 50, 60));
+    }
+
+    @Test
+    void shouldHandleEmptyFlow() throws Exception {
+        // given
+        var flow = Flows.<Flow<?>>empty();
+
+        // when & then
+        assertThat(flow.flattenPar(10).runToList(), Matchers.empty());
+    }
+
+    @Test
+    void shouldHandleSingletonFlow() throws Exception {
+        // given
+        var flow = Flows.fromValues(Flows.fromValues(10));
+
+        // when & then
+        List<Integer> objects = flow.flattenPar(10).runToList();
+        assertThat(objects, contains(10));
+    }
+
+    @Test
+    void shouldNotFlattenNestedFlows() throws Exception {
+        // given
+        var flow = Flows.fromValues(Flows.fromValues(Flows.fromValues(10)));
+
+        // when
+        List<Flow<Integer>> flows = flow.flattenPar(10).runToList();
+
+        // then
+        List<Integer> result = new ArrayList<>();
+        for (Flow<Integer> f : flows) {
+            List<Integer> integers = f.runToList();
+            result.addAll(integers);
+        }
+        assertThat(result, contains(10));
+    }
+
+    @Test
+    void shouldHandleSubsequentFlattenCalls() throws Exception {
+        // given
+        var flow = Flows.fromValues(Flows.fromValues(Flows.fromValues(10), Flows.fromValues(20)));
+
+        // when & then
+        var result = flow.flattenPar(10)
+                .runToList().stream().flatMap(f -> {
+                    try {
+                        return f.runToList().stream();
+                    } catch (Exception e) {
+                        throw new RuntimeException(e);
+                    }
+                })
+                .toList();
+
+        assertThat(result, containsInAnyOrder(10, 20));
+    }
+
+    @Test
+    void shouldRunAtMostParallelismChildFlows() throws Exception {
+        // given
+        var flow = Flows.fromValues(
+                Flows.timeout(Duration.ofMillis(200)).concat(Flows.fromValues(10)),
+                Flows.timeout(Duration.ofMillis(100)).concat(Flows.fromValues(20, 30)),
+                Flows.fromValues(40, 50, 60)
+        );
+
+        // when & then
+        // only one flow can run at a time
+        assertThat(flow.flattenPar(1).runToList(), contains(10, 20, 30, 40, 50, 60));
+        // when parallelism is increased, all flows are run concurrently
+        assertThat(flow.flattenPar(3).runToList(), contains(40, 50, 60, 20, 30, 10));
+    }
+}
diff --git a/flows/src/test/java/com/softwaremill/jox/flows/FlowTest.java b/flows/src/test/java/com/softwaremill/jox/flows/FlowTest.java
index 785aceb..a87d891 100644
--- a/flows/src/test/java/com/softwaremill/jox/flows/FlowTest.java
+++ b/flows/src/test/java/com/softwaremill/jox/flows/FlowTest.java
@@ -14,8 +14,10 @@ import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Optional;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.stream.IntStream;
 
 import com.softwaremill.jox.ChannelError;
 import com.softwaremill.jox.Source;
@@ -588,4 +590,182 @@ class FlowTest {
             return null;
         });
     }
+
+    @Test
+    void scan_shouldScanEmptyFlow() throws Exception {
+        // given
+        Flow<Integer> flow = Flows.empty();
+
+        // when & then
+        Flow<Integer> scannedFlow = flow.scan(0, Integer::sum);
+        assertEquals(List.of(0), scannedFlow.runToList());
+    }
+
+    @Test
+    void scan_shouldScanFlowOfSummedInt() throws Exception {
+        // given
+        Flow<Integer> flow = Flows.fromValues(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
+        Flow<Integer> scannedFlow = flow.scan(0, Integer::sum);
+
+        // when & then
+        assertEquals(List.of(0, 1, 3, 6, 10, 15, 21, 28, 36, 45, 55), scannedFlow.runToList());
+    }
+
+    @Test
+    void scan_shouldScanFlowOfMultipliedInt() throws Exception {
+        // given
+        Flow<Integer> flow = Flows.fromValues(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
+
+        // when & then
+        Flow<Integer> scannedFlow = flow.scan(1, (acc, el) -> acc * el);
+        assertEquals(List.of(1, 1, 2, 6, 24, 120, 720, 5040, 40320, 362880, 3628800), scannedFlow.runToList());
+    }
+
+    @Test
+    void scan_shouldScanFlowOfConcatenatedString() throws Exception {
+        // given
+        Flow<String> flow = Flows.fromValues("f", "l", "o", "w");
+        Flow<String> scannedFlow = flow.scan("my", (acc, el) -> acc + el);
+
+        // when & then
+        assertEquals(List.of("my", "myf", "myfl", "myflo", "myflow"), scannedFlow.runToList());
+    }
+
+    @Test
+    void debounceBy_shouldNotDebounceIfAppliedOnEmptyFlow() throws Exception {
+        // given
+        Flow<Integer> c = Flows.empty();
+
+        // when & then
+        Flow<Integer> s = c.debounceBy(x -> x * 2);
+        assertEquals(List.of(), s.runToList());
+    }
+
+    @Test
+    void debounceBy_shouldNotDebounceIfAppliedOnFlowContainingOnlyDistinctFValue() throws Exception {
+        // given
+        Flow<Integer> c = Flows.fromValues(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
+
+        // when & then
+        Flow<Integer> s = c.debounceBy(x -> x * 2);
+        assertEquals(List.of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10), s.runToList());
+    }
+
+    @Test
+    void debounceBy_shouldDebounceIfAppliedOnFlowContainingRepeatingFValue() throws Exception {
+        // given
+        Flow<Integer> c = Flows.fromValues(1, 1, 2, 3, 4, 4, 5);
+        Flow<Integer> s = c.debounceBy(x -> x * 2);
+
+        // when & then
+        assertEquals(List.of(1, 2, 3, 4, 5), s.runToList());
+    }
+
+    @Test
+    void debounceBy_shouldDebounceSubsequentOddPrimeNumbers() throws Exception {
+        // given
+        Flow<Integer> c = Flows.fromValues(1, 1, 1, 2, 4, 3, 7, 4, 5);
+
+        // when & then
+        Flow<Integer> s = c.debounceBy(x -> x % 2 == 0);
+        assertEquals(List.of(1, 2, 3, 4, 5), s.runToList());
+    }
+
+    @Test
+    void debounce_shouldNotDebounceIfAppliedOnEmptyFlow() throws Exception {
+        // given
+        Flow<Integer> c = Flows.empty();
+
+        // when & then
+        Flow<Integer> s = c.debounce();
+        assertEquals(List.of(), s.runToList());
+    }
+
+    @Test
+    void debounce_shouldNotDebounceIfAppliedOnFlowContainingOnlyDistinctValues() throws Exception {
+        // given
+        Flow<Integer> c = Flows.fromValues(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
+
+        // when & then
+        Flow<Integer> s = c.debounce();
+        assertEquals(List.of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10), s.runToList());
+    }
+
+    @Test
+    void debounce_shouldDebounceIfAppliedOnFlowContainingOnlyRepeatingValues() throws Exception {
+        // given
+        Flow<Integer> c = Flows.fromValues(1, 1, 1, 1, 1);
+
+        // when & then
+        Flow<Integer> s = c.debounce();
+        assertEquals(List.of(1), s.runToList());
+    }
+
+    @Test
+    void debounce_shouldDebounceIfAppliedOnFlowContainingRepeatingElements() throws Exception {
+        // given
+        Flow<Integer> c = Flows.fromValues(1, 1, 2, 3, 4, 4, 5);
+
+        // when & then
+        Flow<Integer> s = c.debounce();
+        assertEquals(List.of(1, 2, 3, 4, 5), s.runToList());
+    }
+
+    @Test
+    void sample_shouldNotSampleAnythingFromEmptyFlow() throws Exception {
+        // given
+        Flow<Integer> c = Flows.empty();
+
+        // when & then
+        Flow<Integer> s = c.sample(5);
+        assertEquals(Collections.emptyList(), s.runToList());
+    }
+
+    @Test
+    void sample_shouldNotSampleAnythingWhenNIsZero() throws Exception {
+        // given
+        Flow<Integer> c = Flows.fromValues(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
+
+        // when & then
+        Flow<Integer> s = c.sample(0);
+        assertEquals(Collections.emptyList(), s.runToList());
+    }
+
+    @Test
+    void sample_shouldSampleEveryElementWhenNIsOne() throws Exception {
+        // given
+        Flow<Integer> c = Flows.fromValues(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
+        int n = 1;
+
+        // when & then
+        Flow<Integer> s = c.sample(n);
+        assertEquals(IntStream.rangeClosed(n, 10).boxed().toList(), s.runToList());
+    }
+
+    @Test
+    void sample_shouldSampleEveryNthElement() throws Exception {
+        // given
+        Flow<Integer> c = Flows.fromValues(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
+        int n = 3;
+
+        // when & then
+        Flow<Integer> s = c.sample(n);
+        assertEquals(List.of(3, 6, 9), s.runToList());
+    }
+
+    @Test
+    void shouldCollectOverSource() throws Exception {
+        // given
+        Flow<Integer> c = Flows.fromValues(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
+
+        // when & then
+        Flow<Integer> s = c.collect(i -> {
+            if (i % 2 == 0) {
+                return Optional.of(i * 10);
+            }
+            return Optional.empty();
+        });
+        List<Integer> result = s.runToList();
+        assertEquals(List.of(20, 40, 60, 80, 100), result);
+    }
 }
diff --git a/flows/src/test/java/com/softwaremill/jox/flows/FlowZipTest.java b/flows/src/test/java/com/softwaremill/jox/flows/FlowZipTest.java
new file mode 100644
index 0000000..09140a4
--- /dev/null
+++ b/flows/src/test/java/com/softwaremill/jox/flows/FlowZipTest.java
@@ -0,0 +1,119 @@
+package com.softwaremill.jox.flows;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import org.junit.jupiter.api.Test;
+import scala.Int;
+
+public class FlowZipTest {
+
+    @Test
+    void shouldNotZipAnythingFromEmptyFlow() throws Exception {
+        // given
+        Flow<Integer> c = Flows.empty();
+        Flow<Map.Entry<Integer, Long>> s = c.zipWithIndex();
+
+        // when & then
+        assertEquals(Collections.emptyList(), s.runToList());
+    }
+
+    @Test
+    void shouldZipFlowWithIndex() throws Exception {
+        // given
+        Flow<Integer> c = Flows.fromValues(1, 2, 3, 4, 5);
+        List<Map.Entry<Integer, Long>> expected = Arrays.asList(
+                Map.entry(1, 0L), Map.entry(2, 1L), Map.entry(3, 2L), Map.entry(4, 3L), Map.entry(5, 4L)
+        );
+
+        // when & then
+        Flow<Map.Entry<Integer, Long>> s = c.zipWithIndex();
+        assertEquals(expected, s.runToList());
+    }
+
+    @Test
+    void zipAll_shouldNotEmitAnyElementWhenBothFlowsAreEmpty() throws Exception {
+        // given
+        Flow<Integer> s = Flows.empty();
+        Flow<String> other = Flows.empty();
+
+        // when & then
+        List<Map.Entry<Integer, String>> result = s.zipAll(other, -1, "foo").runToList();
+        assertEquals(Collections.emptyList(), result);
+    }
+
+    @Test
+    void zipAll_shouldEmitThisElementWhenOtherFlowIsEmpty() throws Exception {
+        // given
+        Flow<Integer> s = Flows.fromValues(1);
+        Flow<String> other = Flows.empty();
+
+        // when & then
+        List<Map.Entry<Integer, String>> result = s.zipAll(other, -1, "foo").runToList();
+        assertEquals(List.of(Map.entry(1, "foo")), result);
+    }
+
+    @Test
+    void zipAll_shouldEmitOtherElementWhenThisFlowIsEmpty() throws Exception {
+        // given
+        Flow<Integer> s = Flows.empty();
+        Flow<String> other = Flows.fromValues("a");
+
+        // when & then
+        List<Map.Entry<Integer, String>> result = s.zipAll(other, -1, "foo").runToList();
+        assertEquals(List.of(Map.entry(-1, "a")), result);
+    }
+
+    @Test
+    void zipAll_shouldEmitMatchingElementsWhenBothFlowsAreOfTheSameSize() throws Exception {
+        // given
+        Flow<Integer> s = Flows.fromValues(1, 2);
+        Flow<String> other = Flows.fromValues("a", "b");
+
+        // when & then
+        List<Map.Entry<Integer, String>> result = s.zipAll(other, -1, "foo").runToList();
+        assertEquals(List.of(Map.entry(1, "a"), Map.entry(2, "b")), result);
+    }
+
+    @Test
+    void zipAll_shouldEmitDefaultForOtherFlowIfThisFlowIsLonger() throws Exception {
+        // given
+        Flow<Integer> s = Flows.fromValues(1, 2, 3);
+        Flow<String> other = Flows.fromValues("a");
+
+        // when & then
+        List<Map.Entry<Integer, String>> result = s.zipAll(other, -1, "foo").runToList();
+        assertEquals(List.of(Map.entry(1, "a"), Map.entry(2, "foo"), Map.entry(3, "foo")), result);
+    }
+
+    @Test
+    void zipAll_shouldEmitDefaultForThisFlowIfOtherFlowIsLonger() throws Exception {
+        // given
+        Flow<Integer> s = Flows.fromValues(1);
+        Flow<String> other = Flows.fromValues("a", "b", "c");
+
+        // when & then
+        List<Map.Entry<Integer, String>> result = s.zipAll(other, -1, "foo").runToList();
+        assertEquals(List.of(Map.entry(1, "a"), Map.entry(-1, "b"), Map.entry(-1, "c")), result);
+    }
+
+    @Test
+    void zip_shouldZipTwoSources() throws Exception {
+        // given
+        Flow<Integer> c1 = Flows.fromValues(1, 2, 3, 0);
+        Flow<Integer> c2 = Flows.fromValues(4, 5, 6);
+        List<Map.Entry<Integer, Integer>> expected = List.of(
+                Map.entry(1, 4),
+                Map.entry(2, 5),
+                Map.entry(3, 6)
+        );
+
+        // when & then
+        List<Map.Entry<Integer, Integer>> s = c1.zip(c2).runToList();
+        assertEquals(expected, s);
+    }
+}
diff --git a/flows/src/test/java/com/softwaremill/jox/flows/FlowsTest.java b/flows/src/test/java/com/softwaremill/jox/flows/FlowsTest.java
index c02d3f9..ec882c9 100644
--- a/flows/src/test/java/com/softwaremill/jox/flows/FlowsTest.java
+++ b/flows/src/test/java/com/softwaremill/jox/flows/FlowsTest.java
@@ -404,6 +404,24 @@ class FlowsTest {
         assertEquals(List.of(0, 1, 2), c.runToList());
     }
 
+    @Test
+    void shouldRunFromIteratorOnlyOnce() throws Exception {
+        Flow<Integer> flow = Flows.fromIterator(List.of(1, 2, 3).iterator());
+
+        assertEquals(List.of(1, 2, 3), flow.runToList()); // first run traverses iterator
+        assertEquals(Collections.emptyList(), flow.runToList()); // second run is empty, as iterator is exhausted
+    }
+
+    @Test
+    void shouldRunFromIteratorSupplierMultipleTimes() throws Exception {
+        List<Integer> source = List.of(1, 2, 3);
+        var flow = Flows.fromIterator(source::iterator);
+
+        for (int i = 0; i < 5; i++) {
+            assertEquals(List.of(1, 2, 3), flow.runToList()); // each run gets new iterator, and is able to traverse it
+        }
+    }
+
     private List<String> toStrings(Flow<ByteChunk> source) throws Exception {
         return source.runToList().stream()
                 .map(chunk -> chunk.convertToString(StandardCharsets.UTF_8))