refactor: replace Tuple2 with Extraction and increase coverage

Author: jponge

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

@@ -9,7 +9,6 @@
 
 import io.smallrye.common.annotation.CheckReturnValue;
 import io.smallrye.mutiny.Multi;
-import io.smallrye.mutiny.tuples.Tuple2;
 
 /**
  * A Gatherer operator transforms a stream of items by accumulating them into an accumulator and extracting
@@ -42,10 +41,11 @@ public interface Gatherer<I, ACC, O> {
      *
      * @param accumulator the current accumulator
      * @param upstreamCompleted whether the upstream has completed
-     * @return an Optional containing a Tuple2 with the updated accumulator and the extracted item, or an empty Optional if no
+     * @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<Tuple2<ACC, O>> extract(ACC accumulator, boolean upstreamCompleted);
+    Optional<Extraction<ACC, O>> extract(ACC accumulator, boolean upstreamCompleted);
 
     /**
      * Finalizes the accumulator and extracts the final item, if any.
@@ -56,6 +56,30 @@ public interface Gatherer<I, ACC, O> {
      */
     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}.
      *
@@ -130,15 +154,15 @@ private ExtractStep(Supplier<ACC> initialAccumulatorSupplier, BiFunction<ACC, I,
          * 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 Tuple2} containing the new accumulator and the value to emit.
+         * 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 Tuple2} with the new accumulator and the value to emit
+         *        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<Tuple2<ACC, O>>> extractor) {
+        public <O> FinalizerStep<I, ACC, O> extract(BiFunction<ACC, Boolean, Optional<Extraction<ACC, O>>> extractor) {
             nonNull(extractor, "extractor");
             return new FinalizerStep<>(initialAccumulatorSupplier, accumulator, extractor);
         }
@@ -154,11 +178,11 @@ public <O> FinalizerStep<I, ACC, O> extract(BiFunction<ACC, Boolean, Optional<Tu
     class FinalizerStep<I, ACC, O> {
         private final Supplier<ACC> initialAccumulatorSupplier;
         private final BiFunction<ACC, I, ACC> accumulator;
-        private final BiFunction<ACC, Boolean, Optional<Tuple2<ACC, O>>> extractor;
+        private final BiFunction<ACC, Boolean, Optional<Extraction<ACC, O>>> extractor;
 
         private FinalizerStep(Supplier<ACC> initialAccumulatorSupplier,
                 BiFunction<ACC, I, ACC> accumulator,
-                BiFunction<ACC, Boolean, Optional<Tuple2<ACC, O>>> extractor) {
+                BiFunction<ACC, Boolean, Optional<Extraction<ACC, O>>> extractor) {
             this.initialAccumulatorSupplier = initialAccumulatorSupplier;
             this.accumulator = accumulator;
             this.extractor = extractor;

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

@@ -8,10 +8,11 @@
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
 
-import io.smallrye.mutiny.tuples.Tuple2;
+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.
  */
 public interface Gatherers {
@@ -30,7 +31,7 @@ public interface Gatherers {
      */
     static <I, ACC, O> Gatherer<I, ACC, O> of(Supplier<ACC> initialAccumulatorSupplier,
             BiFunction<ACC, I, ACC> accumulatorFunction,
-            BiFunction<ACC, Boolean, Optional<Tuple2<ACC, O>>> extractor,
+            BiFunction<ACC, Boolean, Optional<Extraction<ACC, O>>> extractor,
             Function<ACC, Optional<O>> finalizer) {
         return new DefaultGatherer<>(initialAccumulatorSupplier, accumulatorFunction, extractor, finalizer);
     }
@@ -49,7 +50,7 @@ static <I, ACC, O> Gatherer<I, ACC, O> of(Supplier<ACC> initialAccumulatorSuppli
      */
     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(Tuple2.of(acc, acc)), Optional::of);
+                (acc, done) -> done ? Optional.empty() : Optional.of(Extraction.of(acc, acc)), Optional::of);
     }
 
     /**
@@ -84,7 +85,7 @@ static <I> Gatherer<I, List<I>, List<I>> window(int size) {
             return acc;
         }, (acc, completed) -> {
             if (acc.size() == size) {
-                return Optional.of(Tuple2.of(new ArrayList<>(), new ArrayList<>(acc)));
+                return Optional.of(Extraction.of(new ArrayList<>(), new ArrayList<>(acc)));
             }
             return Optional.empty();
         }, acc -> acc.isEmpty()
@@ -109,7 +110,7 @@ static <I> Gatherer<I, List<I>, List<I>> slidingWindow(int size) {
             return acc;
         }, (acc, completed) -> {
             if (acc.size() == size) {
-                return Optional.of(Tuple2.of(acc.stream().skip(1).collect(Collectors.toList()), new ArrayList<>(acc)));
+                return Optional.of(Extraction.of(acc.stream().skip(1).collect(Collectors.toList()), new ArrayList<>(acc)));
             }
             return Optional.empty();
         }, acc -> acc.isEmpty()
@@ -128,12 +129,12 @@ 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<Tuple2<ACC, O>>> extractor;
+        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<Tuple2<ACC, O>>> extractor,
+                BiFunction<ACC, Boolean, Optional<Extraction<ACC, O>>> extractor,
                 Function<ACC, Optional<O>> finalizer) {
             this.initialAccumulatorSupplier = initialAccumulatorSupplier;
             this.accumulatorFunction = accumulatorFunction;
@@ -152,7 +153,7 @@ public ACC accumulate(ACC accumulator, I item) {
         }
 
         @Override
-        public Optional<Tuple2<ACC, O>> extract(ACC accumulator, boolean upstreamCompleted) {
+        public Optional<Extraction<ACC, O>> extract(ACC accumulator, boolean upstreamCompleted) {
             return extractor.apply(accumulator, upstreamCompleted);
         }
 
@@ -171,5 +172,4 @@ public Optional<O> finalize(ACC accumulator) {
     static <I> Gatherer.Builder<I> builder() {
         return new Gatherer.Builder<>();
     }
-
 }

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

@@ -10,8 +10,8 @@
 import io.smallrye.common.annotation.CheckReturnValue;
 import io.smallrye.common.annotation.Experimental;
 import io.smallrye.mutiny.Multi;
+import io.smallrye.mutiny.groups.Gatherer.Extraction;
 import io.smallrye.mutiny.operators.multi.MultiGather;
-import io.smallrye.mutiny.tuples.Tuple2;
 
 /**
  * A builder to gather items emitted by a {@link Multi} into an accumulator.
@@ -97,15 +97,15 @@ private ExtractStep(Multi<I> upstream, Supplier<ACC> initialAccumulatorSupplier,
          * 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 Tuple2} containing the new accumulator and the value to emit.
+         * 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 Tuple2} with the new accumulator and the value to emit
+         *        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<Tuple2<ACC, O>>> extractor) {
+        public <O> FinalizerStep<I, ACC, O> extract(BiFunction<ACC, Boolean, Optional<Extraction<ACC, O>>> extractor) {
             nonNull(extractor, "extractor");
             return new FinalizerStep<>(upstream, initialAccumulatorSupplier, accumulator, extractor);
         }
@@ -122,12 +122,12 @@ public static class FinalizerStep<I, ACC, O> {
         private final Multi<I> upstream;
         private final Supplier<ACC> initialAccumulatorSupplier;
         private final BiFunction<ACC, I, ACC> accumulator;
-        private final BiFunction<ACC, Boolean, Optional<Tuple2<ACC, O>>> extractor;
+        private final BiFunction<ACC, Boolean, Optional<Extraction<ACC, O>>> extractor;
 
         private FinalizerStep(Multi<I> upstream,
                 Supplier<ACC> initialAccumulatorSupplier,
                 BiFunction<ACC, I, ACC> accumulator,
-                BiFunction<ACC, Boolean, Optional<Tuple2<ACC, O>>> extractor) {
+                BiFunction<ACC, Boolean, Optional<Extraction<ACC, O>>> extractor) {
             this.upstream = upstream;
             this.initialAccumulatorSupplier = initialAccumulatorSupplier;
             this.accumulator = accumulator;

implementation/src/main/java/io/smallrye/mutiny/operators/multi/MultiGather.java

@@ -7,9 +7,9 @@
 
 import io.smallrye.mutiny.Multi;
 import io.smallrye.mutiny.groups.Gatherer;
+import io.smallrye.mutiny.groups.Gatherer.Extraction;
 import io.smallrye.mutiny.helpers.Subscriptions;
 import io.smallrye.mutiny.subscription.MultiSubscriber;
-import io.smallrye.mutiny.tuples.Tuple2;
 
 public class MultiGather<I, ACC, O> extends AbstractMultiOperator<I, O> {
 
@@ -77,14 +77,14 @@ public void onItem(I item) {
                 if (acc == null) {
                     throw new NullPointerException("The accumulator returned a null value");
                 }
-                Optional<Tuple2<ACC, O>> mapping = gatherer.extract(acc, false);
+                Optional<Extraction<ACC, O>> mapping = gatherer.extract(acc, false);
                 if (mapping == null) {
                     throw new NullPointerException("The extractor returned a null value");
                 }
                 if (mapping.isPresent()) {
-                    Tuple2<ACC, O> tuple = mapping.get();
-                    acc = tuple.getItem1();
-                    O value = tuple.getItem2();
+                    Extraction<ACC, O> result = mapping.get();
+                    acc = result.nextAccumulator();
+                    O value = result.nextItem();
                     if (acc == null) {
                         throw new NullPointerException("The extractor returned a null accumulator value");
                     }
@@ -125,14 +125,14 @@ private void drainRemainingElements() {
                         return;
                     }
                     try {
-                        Optional<Tuple2<ACC, O>> mapping = gatherer.extract(acc, true);
+                        Optional<Extraction<ACC, O>> mapping = gatherer.extract(acc, true);
                         if (mapping == null) {
                             throw new NullPointerException("The extractor returned a null value");
                         }
                         if (mapping.isPresent()) {
-                            Tuple2<ACC, O> tuple = mapping.get();
-                            acc = tuple.getItem1();
-                            O value = tuple.getItem2();
+                            Extraction<ACC, O> result = mapping.get();
+                            acc = result.nextAccumulator();
+                            O value = result.nextItem();
                             if (acc == null) {
                                 throw new NullPointerException("The extractor returned a null accumulator value");
                             }

implementation/src/test/java/io/smallrye/mutiny/operators/multi/MultiGatherTest.java

@@ -11,9 +11,10 @@
 import org.junit.jupiter.api.Test;
 
 import io.smallrye.mutiny.Multi;
+import io.smallrye.mutiny.groups.Gatherer;
+import io.smallrye.mutiny.groups.Gatherer.Extraction;
 import io.smallrye.mutiny.groups.Gatherers;
 import io.smallrye.mutiny.helpers.test.AssertSubscriber;
-import io.smallrye.mutiny.tuples.Tuple2;
 
 class MultiGatherTest {
 
@@ -99,7 +100,7 @@ void gatherLinesOfText() {
                     String str = sb.toString();
                     if (str.contains("\n")) {
                         String[] lines = str.split("\n", 2);
-                        return Optional.of(Tuple2.of(new StringBuilder(lines[1]), lines[0]));
+                        return Optional.of(Extraction.of(new StringBuilder(lines[1]), lines[0]));
                     }
                     return Optional.empty();
                 })
@@ -133,7 +134,7 @@ void checkCompletionCorrectness() {
                     String str = sb.toString();
                     if (str.contains(",")) {
                         String[] lines = str.split(",", 2);
-                        return Optional.of(Tuple2.of(new StringBuilder(lines[1]), lines[0]));
+                        return Optional.of(Extraction.of(new StringBuilder(lines[1]), lines[0]));
                     }
                     return Optional.empty();
                 })
@@ -164,7 +165,8 @@ void rejectNullParamsInApi() {
                 .withMessageContaining("accumulator");
         assertThatIllegalArgumentException()
                 .isThrownBy(() -> multi.onItem().gather().into(ArrayList<Integer>::new).accumulate((a, b) -> a)
-                        .extract((BiFunction<ArrayList<Integer>, Boolean, Optional<Tuple2<ArrayList<Integer>, Object>>>) null))
+                        .extract(
+                                (BiFunction<ArrayList<Integer>, Boolean, Optional<Extraction<ArrayList<Integer>, Object>>>) null))
                 .withMessageContaining("extractor");
         assertThatIllegalArgumentException()
                 .isThrownBy(() -> multi.onItem().gather().into(ArrayList<Integer>::new).accumulate((a, b) -> a)
@@ -220,7 +222,7 @@ void rejectNullInExtractorOptionalTupleLeft() {
                     acc.add(next);
                     return acc;
                 })
-                .extract((acc, completed) -> Optional.of(Tuple2.of(null, "ok")))
+                .extract((acc, completed) -> Optional.of(Extraction.of(null, "ok")))
                 .finalize(acc -> Optional.empty())
                 .subscribe().withSubscriber(AssertSubscriber.create(Long.MAX_VALUE));
         sub.assertFailedWith(NullPointerException.class, "The extractor returned a null accumulator value");
@@ -235,7 +237,7 @@ void rejectNullInExtractorOptionalTupleRight() {
                     acc.add(next);
                     return acc;
                 })
-                .extract((acc, completed) -> Optional.of(Tuple2.of(new ArrayList<>(), null)))
+                .extract((acc, completed) -> Optional.of(Extraction.of(new ArrayList<>(), null)))
                 .finalize(acc -> Optional.empty())
                 .subscribe().withSubscriber(AssertSubscriber.create(Long.MAX_VALUE));
         sub.assertFailedWith(NullPointerException.class, "The extractor returned a null value to emit");
@@ -350,4 +352,43 @@ void rejectBadRequests() {
         sub.request(-10L).assertFailedWith(IllegalArgumentException.class,
                 "The number of items requested must be strictly positive");
     }
+
+    @Test
+    void builderApi() {
+        Gatherer<String, StringBuilder, String> gatherer = Gatherers.<String> builder()
+                .into(StringBuilder::new)
+                .accumulate(StringBuilder::append)
+                .extract((sb, completed) -> {
+                    String str = sb.toString();
+                    if (str.contains("\n")) {
+                        String[] lines = str.split("\n", 2);
+                        return Optional.of(Extraction.of(new StringBuilder(lines[1]), lines[0]));
+                    }
+                    return Optional.empty();
+                })
+                .finalize(sb -> Optional.of(sb.toString()));
+
+        List<String> chunks = List.of(
+                "Hello", " ", "world!\n",
+                "This is a test\n",
+                "==\n==",
+                "\n\nThis", " is", " ", "amazing\n\n");
+        AssertSubscriber<String> sub = Multi.createFrom().iterable(chunks)
+                .onItem().gather(gatherer)
+                .subscribe().withSubscriber(AssertSubscriber.create());
+
+        sub.awaitNextItems(2);
+        assertThat(sub.getItems()).containsExactly("Hello world!", "This is a test");
+
+        sub.request(Long.MAX_VALUE);
+        assertThat(sub.getItems()).containsExactly(
+                "Hello world!",
+                "This is a test",
+                "==",
+                "==",
+                "",
+                "This is amazing",
+                "",
+                "");
+    }
 }

reactive-streams-tck-tests/src/test/java/io/smallrye/mutiny/tcktests/MultiGatherTckTest.java

@@ -6,7 +6,7 @@
 import java.util.stream.LongStream;
 
 import io.smallrye.mutiny.Multi;
-import io.smallrye.mutiny.tuples.Tuple2;
+import io.smallrye.mutiny.groups.Gatherer.Extraction;
 
 public class MultiGatherTckTest extends AbstractPublisherTck<Long> {
 
@@ -24,7 +24,7 @@ public Flow.Publisher<Long> createFlowPublisher(long elements) {
                     if (list.isEmpty()) {
                         return Optional.empty();
                     } else {
-                        return Optional.of(Tuple2.of(new ArrayList<>(), list.get(0)));
+                        return Optional.of(Extraction.of(new ArrayList<>(), list.get(0)));
                     }
                 })
                 .finalize(list -> Optional.empty());

workshop-examples/src/main/java/_03_composition_transformation/_22_Multi_Chunks_To_Sentence_Gather.java

@@ -10,7 +10,7 @@
 
 import io.smallrye.mutiny.Multi;
 import io.smallrye.mutiny.Uni;
-import io.smallrye.mutiny.tuples.Tuple2;
+import io.smallrye.mutiny.groups.Gatherer.Extraction;
 
 public class _22_Multi_Chunks_To_Sentence_Gather {
 
@@ -34,7 +34,7 @@ public static void main(String[] args) throws InterruptedException {
                     String str = sb.toString();
                     if (str.contains("\n")) {
                         String[] lines = str.split("\n", 2);
-                        return Optional.of(Tuple2.of(new StringBuilder(lines[1]), lines[0]));
+                        return Optional.of(Extraction.of(new StringBuilder(lines[1]), lines[0]));
                     }
                     return Optional.empty();
                 })