Merge pull request #1845 from jponge/feat/multi-gather

feat: introduction of a Multi gathering operator

Author: jponge

implementation/src/main/java/io/smallrye/mutiny/groups/Gatherer.java

@@ -0,0 +1,212 @@
+package io.smallrye.mutiny.groups;
+
+import static io.smallrye.mutiny.helpers.ParameterValidation.nonNull;
+
+import java.util.Optional;
+import java.util.function.BiFunction;
+import java.util.function.Function;
+import java.util.function.Supplier;
+
+import io.smallrye.common.annotation.CheckReturnValue;
+import io.smallrye.common.annotation.Experimental;
+import io.smallrye.mutiny.Multi;
+
+/**
+ * A Gatherer operator transforms a stream of items by accumulating them into an accumulator and extracting
+ * items from that accumulator when certain conditions are met.
+ *
+ * @param <I> the type of the items emitted by the upstream
+ * @param <ACC> the type of the accumulator
+ * @param <O> the type of the items emitted to the downstream
+ */
+@Experimental("This API is still being designed and may change in the future")
+public interface Gatherer<I, ACC, O> {
+
+    /**
+     * Creates a new accumulator.
+     *
+     * @return a new accumulator
+     */
+    ACC accumulator();
+
+    /**
+     * Accumulates an item into the accumulator.
+     *
+     * @param accumulator the current accumulator
+     * @param item the item to accumulate
+     * @return the updated accumulator
+     */
+    ACC accumulate(ACC accumulator, I item);
+
+    /**
+     * Extracts an item from the accumulator.
+     *
+     * @param accumulator the current accumulator
+     * @param upstreamCompleted whether the upstream has completed
+     * @return an Optional containing a Extraction with the updated accumulator and the extracted item, or an empty Optional if
+     *         no
+     *         item can be extracted
+     */
+    Optional<Extraction<ACC, O>> extract(ACC accumulator, boolean upstreamCompleted);
+
+    /**
+     * Finalizes the accumulator and extracts the final item, if any.
+     * This method is called when the upstream has completed and no more items can be extracted using the extract method.
+     *
+     * @param accumulator the current accumulator
+     * @return an Optional containing the final item, or an empty Optional if no final item can be extracted
+     */
+    Optional<O> finalize(ACC accumulator);
+
+    /**
+     * An extraction result containing the next accumulator and the next item to emit.
+     *
+     * @param nextAccumulator the next accumulator
+     * @param nextItem the next item to emit
+     * @param <ACC> the type of the accumulator
+     * @param <O> the type of the item to emit
+     */
+    record Extraction<ACC, O>(ACC nextAccumulator, O nextItem) {
+
+        /**
+         * Creates a new {@link Extraction} instance.
+         *
+         * @param nextAccumulator the next accumulator
+         * @param nextItem the next item to emit
+         * @return a new {@link Extraction} instance
+         * @param <ACC> the type of the accumulator
+         * @param <O> the type of the item to emit
+         */
+        public static <ACC, O> Extraction<ACC, O> of(ACC nextAccumulator, O nextItem) {
+            return new Extraction<>(nextAccumulator, nextItem);
+        }
+    }
+
+    /**
+     * Builder for creating a {@link Gatherer}.
+     *
+     * @param <I> the type of the items emitted by the upstream
+     */
+    class Builder<I> {
+
+        /**
+         * Specifies the initial accumulator supplier.
+         * <p>
+         * The initial accumulator supplier is used to create a new accumulator.
+         *
+         * @param initialAccumulatorSupplier the supplier for the initial accumulator
+         * @param <ACC> the type of the accumulator
+         * @return the next step in the builder
+         */
+        @CheckReturnValue
+        public <ACC> InitialAccumulatorStep<I, ACC> into(Supplier<ACC> initialAccumulatorSupplier) {
+            nonNull(initialAccumulatorSupplier, "initialAccumulatorSupplier");
+            return new InitialAccumulatorStep<>(initialAccumulatorSupplier);
+        }
+    }
+
+    /**
+     * The first step in the builder to gather items emitted by a {@link Multi} into an accumulator.
+     *
+     * @param <I> the type of the items emitted by the upstream {@link Multi}
+     * @param <ACC> the type of the accumulator
+     */
+    class InitialAccumulatorStep<I, ACC> {
+        private final Supplier<ACC> initialAccumulatorSupplier;
+
+        private InitialAccumulatorStep(Supplier<ACC> initialAccumulatorSupplier) {
+            this.initialAccumulatorSupplier = initialAccumulatorSupplier;
+        }
+
+        /**
+         * Specifies the accumulator function.
+         * <p>
+         * The accumulator function is used to accumulate the items emitted by the upstream.
+         *
+         * @param accumulator the accumulator function, which takes the current accumulator and the item emitted by the
+         *        upstream, and returns the new accumulator
+         * @return the next step in the builder
+         */
+        @CheckReturnValue
+        public ExtractStep<I, ACC> accumulate(BiFunction<ACC, I, ACC> accumulator) {
+            nonNull(accumulator, "accumulator");
+            return new ExtractStep<>(initialAccumulatorSupplier, accumulator);
+        }
+    }
+
+    /**
+     * The second step in the builder to gather items emitted by a {@link Multi} into an accumulator.
+     *
+     * @param <I> the type of the items emitted by the upstream {@link Multi}
+     * @param <ACC> the type of the accumulator
+     */
+    class ExtractStep<I, ACC> {
+        private final Supplier<ACC> initialAccumulatorSupplier;
+        private final BiFunction<ACC, I, ACC> accumulator;
+
+        private ExtractStep(Supplier<ACC> initialAccumulatorSupplier, BiFunction<ACC, I, ACC> accumulator) {
+            this.initialAccumulatorSupplier = initialAccumulatorSupplier;
+            this.accumulator = accumulator;
+        }
+
+        /**
+         * Specifies the extractor function.
+         * <p>
+         * The extractor function is used to extract the items from the accumulator.
+         * When the extractor function returns an empty {@link Optional}, no value is emitted.
+         * When the extractor function returns a non-empty {@link Optional}, the value is emitted, and the accumulator is
+         * updated.
+         * This is done by returning a {@link Extraction} containing the new accumulator and the value to emit.
+         *
+         * @param extractor the extractor function, which takes the current accumulator and returns an {@link Optional}
+         *        containing a {@link Extraction} with the new accumulator and the value to emit
+         * @param <O> the type of the value to emit
+         * @return the next step in the builder
+         */
+        @CheckReturnValue
+        public <O> FinalizerStep<I, ACC, O> extract(BiFunction<ACC, Boolean, Optional<Extraction<ACC, O>>> extractor) {
+            nonNull(extractor, "extractor");
+            return new FinalizerStep<>(initialAccumulatorSupplier, accumulator, extractor);
+        }
+    }
+
+    /**
+     * The third step in the builder to gather items emitted by a {@link Multi} into an accumulator.
+     *
+     * @param <I> the type of the items emitted by the upstream
+     * @param <ACC> the type of the accumulator
+     * @param <O> the type of the items emitted to the downstream
+     */
+    class FinalizerStep<I, ACC, O> {
+        private final Supplier<ACC> initialAccumulatorSupplier;
+        private final BiFunction<ACC, I, ACC> accumulator;
+        private final BiFunction<ACC, Boolean, Optional<Extraction<ACC, O>>> extractor;
+
+        private FinalizerStep(Supplier<ACC> initialAccumulatorSupplier,
+                BiFunction<ACC, I, ACC> accumulator,
+                BiFunction<ACC, Boolean, Optional<Extraction<ACC, O>>> extractor) {
+            this.initialAccumulatorSupplier = initialAccumulatorSupplier;
+            this.accumulator = accumulator;
+            this.extractor = extractor;
+        }
+
+        /**
+         * Specifies the finalizer function.
+         * <p>
+         * The finalizer function is used to emit the final value upon completion of the upstream and when there are no more
+         * items that can be extracted from the accumulator.
+         * When the finalizer function returns an empty {@link Optional}, no value is emitted before the completion signal.
+         * When the finalizer function returns a non-empty {@link Optional}, the value is emitted before the completion signal.
+         *
+         * @param finalizer the finalizer function, which takes the current accumulator and returns an {@link Optional}
+         *        containing the value to emit before the completion signal, if any
+         * @return the gathering {@link Multi}
+         */
+        @CheckReturnValue
+        public Gatherer<I, ACC, O> finalize(Function<ACC, Optional<O>> finalizer) {
+            nonNull(finalizer, "finalizer");
+            return Gatherers.of(initialAccumulatorSupplier, accumulator, extractor, finalizer);
+        }
+    }
+
+}

implementation/src/main/java/io/smallrye/mutiny/groups/Gatherers.java

@@ -0,0 +1,177 @@
+package io.smallrye.mutiny.groups;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+import java.util.function.BiFunction;
+import java.util.function.Function;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+
+import io.smallrye.common.annotation.Experimental;
+import io.smallrye.mutiny.groups.Gatherer.Extraction;
+
+/**
+ * Factory interface for creating {@link Gatherer} instances.
+ * <p>
+ * This interface provides various static methods to create different types of gatherers.
+ */
+@Experimental("This API is still being designed and may change in the future")
+public interface Gatherers {
+
+    /**
+     * Creates a new {@link Gatherer} with the specified components.
+     *
+     * @param initialAccumulatorSupplier the supplier for the initial accumulator
+     * @param accumulatorFunction the function to accumulate items into the accumulator
+     * @param extractor the function to extract items from the accumulator
+     * @param finalizer the function to extract the final item from the accumulator
+     * @param <I> the type of the items emitted by the upstream
+     * @param <ACC> the type of the accumulator
+     * @param <O> the type of the items emitted to the downstream
+     * @return a new {@link Gatherer}
+     */
+    static <I, ACC, O> Gatherer<I, ACC, O> of(Supplier<ACC> initialAccumulatorSupplier,
+            BiFunction<ACC, I, ACC> accumulatorFunction,
+            BiFunction<ACC, Boolean, Optional<Extraction<ACC, O>>> extractor,
+            Function<ACC, Optional<O>> finalizer) {
+        return new DefaultGatherer<>(initialAccumulatorSupplier, accumulatorFunction, extractor, finalizer);
+    }
+
+    /**
+     * Creates a new {@link Gatherer} that performs a scan operation.
+     * The scan operation applies a function to each item emitted by the upstream, using the result of the previous
+     * application as the first argument to the function. The initial value is provided by the initialAccumulatorSupplier.
+     * <p>
+     * Each intermediate result is emitted downstream.
+     *
+     * @param initialAccumulatorSupplier the supplier for the initial accumulator
+     * @param accumulatorFunction the function to accumulate items
+     * @param <I> the type of the items emitted by the upstream and downstream
+     * @return a new {@link Gatherer} that performs a scan operation
+     */
+    static <I> Gatherer<I, I, I> scan(Supplier<I> initialAccumulatorSupplier, BiFunction<I, I, I> accumulatorFunction) {
+        return of(initialAccumulatorSupplier, accumulatorFunction,
+                (acc, done) -> done ? Optional.empty() : Optional.of(Extraction.of(acc, acc)), Optional::of);
+    }
+
+    /**
+     * Creates a new {@link Gatherer} that performs a fold operation.
+     * The fold operation applies a function to each item emitted by the upstream, using the result of the previous
+     * application as the first argument to the function. The initial value is provided by the initialAccumulatorSupplier.
+     * <p>
+     * Only emits the final result when the upstream completes.
+     *
+     * @param initialAccumulatorSupplier the supplier for the initial accumulator
+     * @param accumulatorFunction the function to accumulate items
+     * @param <I> the type of the items emitted by the upstream and downstream
+     * @return a new {@link Gatherer} that performs a fold operation
+     */
+    static <I> Gatherer<I, I, I> fold(Supplier<I> initialAccumulatorSupplier, BiFunction<I, I, I> accumulatorFunction) {
+        return of(initialAccumulatorSupplier, accumulatorFunction, (acc, done) -> Optional.empty(), Optional::of);
+    }
+
+    /**
+     * Creates a new {@link Gatherer} that performs a windowing operation.
+     * The windowing operation collects items emitted by the upstream into non-overlapping windows of the specified size.
+     * When a window is full, it is emitted downstream and a new window is started.
+     * If the upstream completes before a window is full, the current window is emitted if it is not empty.
+     *
+     * @param size the size of the window
+     * @param <I> the type of the items emitted by the upstream
+     * @return a new {@link Gatherer} that performs a windowing operation
+     */
+    static <I> Gatherer<I, List<I>, List<I>> window(int size) {
+        return of(ArrayList::new, (acc, next) -> {
+            acc.add(next);
+            return acc;
+        }, (acc, completed) -> {
+            if (acc.size() == size) {
+                return Optional.of(Extraction.of(new ArrayList<>(), new ArrayList<>(acc)));
+            }
+            return Optional.empty();
+        }, acc -> acc.isEmpty()
+                ? Optional.empty()
+                : Optional.of(acc));
+    }
+
+    /**
+     * Creates a new {@link Gatherer} that performs a sliding window operation.
+     * The sliding window operation collects items emitted by the upstream into overlapping windows of the specified size.
+     * When a window is full, it is emitted downstream and a new window is started with all but the first item from the previous
+     * window.
+     * If the upstream completes before a window is full, the current window is emitted if it is not empty.
+     *
+     * @param size the size of the window
+     * @param <I> the type of the items emitted by the upstream
+     * @return a new {@link Gatherer} that performs a sliding window operation
+     */
+    static <I> Gatherer<I, List<I>, List<I>> slidingWindow(int size) {
+        return of(ArrayList::new, (acc, item) -> {
+            acc.add(item);
+            return acc;
+        }, (acc, completed) -> {
+            if (acc.size() == size) {
+                return Optional.of(Extraction.of(acc.stream().skip(1).collect(Collectors.toList()), new ArrayList<>(acc)));
+            }
+            return Optional.empty();
+        }, acc -> acc.isEmpty()
+                ? Optional.empty()
+                : Optional.of(acc));
+    }
+
+    /**
+     * Default implementation of the {@link Gatherer} interface.
+     *
+     * @param <I> the type of the items emitted by the upstream
+     * @param <ACC> the type of the accumulator
+     * @param <O> the type of the items emitted to the downstream
+     */
+    class DefaultGatherer<I, ACC, O> implements Gatherer<I, ACC, O> {
+
+        private final Supplier<ACC> initialAccumulatorSupplier;
+        private final BiFunction<ACC, I, ACC> accumulatorFunction;
+        private final BiFunction<ACC, Boolean, Optional<Extraction<ACC, O>>> extractor;
+        private final Function<ACC, Optional<O>> finalizer;
+
+        public DefaultGatherer(Supplier<ACC> initialAccumulatorSupplier,
+                BiFunction<ACC, I, ACC> accumulatorFunction,
+                BiFunction<ACC, Boolean, Optional<Extraction<ACC, O>>> extractor,
+                Function<ACC, Optional<O>> finalizer) {
+            this.initialAccumulatorSupplier = initialAccumulatorSupplier;
+            this.accumulatorFunction = accumulatorFunction;
+            this.extractor = extractor;
+            this.finalizer = finalizer;
+        }
+
+        @Override
+        public ACC accumulator() {
+            return initialAccumulatorSupplier.get();
+        }
+
+        @Override
+        public ACC accumulate(ACC accumulator, I item) {
+            return accumulatorFunction.apply(accumulator, item);
+        }
+
+        @Override
+        public Optional<Extraction<ACC, O>> extract(ACC accumulator, boolean upstreamCompleted) {
+            return extractor.apply(accumulator, upstreamCompleted);
+        }
+
+        @Override
+        public Optional<O> finalize(ACC accumulator) {
+            return finalizer.apply(accumulator);
+        }
+    }
+
+    /**
+     * Creates a new {@link Gatherer} builder.
+     *
+     * @param <I> the type of the items emitted by the upstream
+     * @return the builder
+     */
+    static <I> Gatherer.Builder<I> builder() {
+        return new Gatherer.Builder<>();
+    }
+}