BAEL-9210: Stream Gatherers (#18513)

Author: brokenhardisk

core-java-modules/core-java-streams-7/pom.xml

@@ -35,8 +35,8 @@
     </build>
 
     <properties>
-        <maven.compiler.source>12</maven.compiler.source>
-        <maven.compiler.target>12</maven.compiler.target>
+        <maven.compiler.source>24</maven.compiler.source>
+        <maven.compiler.target>24</maven.compiler.target>
         <vavr.version>0.10.2</vavr.version>
         <assertj-core.version>3.23.1</assertj-core.version>
         <commons-lang3.version>3.12.0</commons-lang3.version>

core-java-modules/core-java-streams-7/src/main/java/com/baeldung/streams/gatherer/NumericSumGatherer.java

@@ -0,0 +1,39 @@
+package com.baeldung.streams.gatherer;
+
+import java.util.ArrayList;
+import java.util.function.BiConsumer;
+import java.util.function.Supplier;
+import java.util.stream.Gatherer;
+
+public class NumericSumGatherer implements Gatherer<Integer, ArrayList<Integer>, Integer> {
+
+    @Override
+    public Supplier<ArrayList<Integer>> initializer() {
+        return ArrayList::new;
+    }
+
+    @Override
+    public Integrator<ArrayList<Integer>, Integer, Integer> integrator() {
+        return new Integrator<>() {
+            @Override
+            public boolean integrate(ArrayList<Integer> state, Integer element, Downstream<? super Integer> downstream) {
+                if (state.isEmpty()) {
+                    state.add(element);
+                } else {
+                    state.addFirst(state.getFirst() + element);
+                }
+                return true;
+            }
+        };
+    }
+
+    @Override
+    public BiConsumer<ArrayList<Integer>, Downstream<? super Integer>> finisher() {
+        return (state, downstream) -> {
+            if (!downstream.isRejecting() && !state.isEmpty()) {
+                downstream.push(state.getFirst());
+                state.clear();
+            }
+        };
+    }
+}

core-java-modules/core-java-streams-7/src/main/java/com/baeldung/streams/gatherer/SentenceSplitterGatherer.java

@@ -0,0 +1,36 @@
+package com.baeldung.streams.gatherer;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.function.BinaryOperator;
+import java.util.function.Supplier;
+import java.util.stream.Gatherer;
+
+public class SentenceSplitterGatherer implements Gatherer<String, List<String>,String> {
+
+    @Override
+    public Supplier<List<String>> initializer() {
+        return ArrayList::new;
+    }
+
+    @Override
+    public BinaryOperator<List<String>> combiner() {
+        return (left, right) -> {
+            left.addAll(right);
+            return left;
+        };
+    }
+
+    @Override
+    public Integrator<List<String>, String, String> integrator() {
+        return (state, element, downstream) -> {
+            var words = element.split("\\s+");
+            for (var word : words) {
+                state.add(word);
+                downstream.push(word);
+            }
+            return true;
+        };
+    }
+}

core-java-modules/core-java-streams-7/src/main/java/com/baeldung/streams/gatherer/SlidingWindowGatherer.java

@@ -0,0 +1,40 @@
+package com.baeldung.streams.gatherer;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.function.BiConsumer;
+import java.util.function.Supplier;
+import java.util.stream.Gatherer;
+
+public class SlidingWindowGatherer implements Gatherer<Integer, ArrayList<Integer>, List<Integer>> {
+
+    @Override
+    public Supplier<ArrayList<Integer>> initializer() {
+        return ArrayList::new;
+    }
+
+    @Override
+    public Integrator<ArrayList<Integer>, Integer, List<Integer>> integrator() {
+        return new Integrator<>() {
+            @Override
+            public boolean integrate(ArrayList<Integer> state, Integer element, Downstream<? super List<Integer>> downstream) {
+                state.add(element);
+                if (state.size() == 3) {
+                    downstream.push(new ArrayList<>(state));
+                    state.removeFirst();
+                }
+                return true;
+            }
+        };
+    }
+
+    @Override
+    public BiConsumer<ArrayList<Integer>, Downstream<? super List<Integer>>> finisher() {
+        return (state, downstream) -> {
+            if (state.size()==3) {
+                downstream.push(new ArrayList<>(state));
+            }
+        };
+
+    }
+}

core-java-modules/core-java-streams-7/src/test/java/com/baeldung/streams/gatherer/GathererUnitTest.java

@@ -0,0 +1,62 @@
+package com.baeldung.streams.gatherer;
+
+import java.util.List;
+import java.util.stream.Gatherer;
+import java.util.stream.Gatherers;
+import java.util.stream.Stream;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+class GathererUnitTest {
+
+    @Test
+    void givenNumbers_whenFolded_thenSumIsEmitted() {
+        Stream<Integer> numbers = Stream.of(1, 2, 3, 4, 5);
+        Stream<Integer> folded = numbers.gather(Gatherers.fold(() -> 0, Integer::sum));
+        List<Integer> resultList = folded.toList();
+        Assertions.assertEquals(1, resultList.size());
+        Assertions.assertEquals(Integer.valueOf(15), resultList.getFirst());
+    }
+
+    @Test
+    void givenWords_whenMappedConcurrently_thenUppercasedWordsAreEmitted() {
+        Stream<String> words = Stream.of("a", "b", "c", "d");
+        List<String> resultList = words.gather(Gatherers.mapConcurrent(2, String::toUpperCase))
+            .toList();
+        Assertions.assertEquals(4, resultList.size());
+        Assertions.assertEquals(List.of("A", "B", "C", "D"), resultList);
+    }
+
+    @Test
+    void givenNumbers_whenScanned_thenRunningTotalsAreEmitted() {
+        Stream<Integer> numbers = Stream.of(1, 2, 3, 4);
+        List<Integer> resultList = numbers.gather(Gatherers.scan(() -> 0, Integer::sum))
+            .toList();
+        Assertions.assertEquals(4, resultList.size());
+        Assertions.assertEquals(List.of(1, 3, 6, 10), resultList);
+    }
+
+    @Test
+    void givenNumbers_whenWindowedSliding_thenOverlappingWindowsAreEmitted() {
+        List<List<Integer>> expectedOutput = List.of(List.of(1, 2, 3), List.of(2, 3, 4), List.of(3, 4, 5));
+        Stream<Integer> numbers = Stream.of(1, 2, 3, 4, 5);
+        List<List<Integer>> resultList = numbers.gather(Gatherers.windowSliding(3))
+            .toList();
+        Assertions.assertEquals(3, resultList.size());
+        Assertions.assertEquals(expectedOutput, resultList);
+    }
+
+    @Test
+    void givenStrings_whenUsingCustomGatherer_thenLengthsAreCalculated() {
+        List<Integer> expectedOutput = List.of(5, 6, 3);
+        Stream<String> inputStrings = Stream.of("apple", "banana", "cat");
+        List<Object> resultList = inputStrings.gather(Gatherer.of((state, element, downstream) -> {
+                downstream.push(element.length());
+                return true;
+            }))
+            .toList();
+        Assertions.assertEquals(3, resultList.size());
+        Assertions.assertEquals(expectedOutput, resultList);
+    }
+}

core-java-modules/core-java-streams-7/src/test/java/com/baeldung/streams/gatherer/NumericSumGathererUnitTest.java

@@ -0,0 +1,18 @@
+package com.baeldung.streams.gatherer;
+
+import java.util.List;
+import java.util.stream.Stream;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+class NumericSumGathererUnitTest {
+
+    @Test
+    void givenNumbers_whenUsingCustomManyToOneGatherer_thenSumIsCalculated() {
+        Stream<Integer> inputValues = Stream.of(1, 2, 3, 4, 5, 6);
+        List<Integer> result = inputValues.gather(new NumericSumGatherer())
+            .toList();
+        Assertions.assertEquals(Integer.valueOf(21), result.getFirst());
+    }
+}

core-java-modules/core-java-streams-7/src/test/java/com/baeldung/streams/gatherer/SentenceSplitterGathererUnitTest.java

@@ -0,0 +1,19 @@
+package com.baeldung.streams.gatherer;
+
+import java.util.List;
+import java.util.stream.Stream;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+class SentenceSplitterGathererUnitTest {
+
+    @Test
+    void givenSentences_whenUsingCustomOneToManyGatherer_thenWordsAreExtracted() {
+        List<String> expectedOutput = List.of("hello", "world", "java", "streams");
+        Stream<String> sentences = Stream.of("hello world", "java streams");
+        List<String> words = sentences.gather(new SentenceSplitterGatherer())
+            .toList();
+        Assertions.assertEquals(expectedOutput, words);
+    }
+}

core-java-modules/core-java-streams-7/src/test/java/com/baeldung/streams/gatherer/SlidingWindowGathererUnitTest.java

@@ -0,0 +1,21 @@
+package com.baeldung.streams.gatherer;
+
+import java.util.List;
+import java.util.stream.Stream;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+class SlidingWindowGathererUnitTest {
+
+    @Test
+    void givenNumbers_whenWindowedSliding_thenOverlappingWindowsAreEmitted() {
+        List<List<Integer>> expectedOutput = List.of(List.of(1, 2, 3), List.of(2, 3, 4), List.of(3, 4, 5));
+        Stream<Integer> numbers = Stream.of(1, 2, 3, 4, 5);
+        List<List<Integer>> resultList = numbers.gather(new SlidingWindowGatherer())
+            .toList();
+        Assertions.assertEquals(3, resultList.size());
+        Assertions.assertEquals(expectedOutput, resultList);
+    }
+
+}

core-java-modules/pom.xml

@@ -60,7 +60,7 @@
         <module>core-java-streams-4</module>
         <module>core-java-streams-5</module>
         <module>core-java-streams-6</module>
-        <module>core-java-streams-7</module>
+        <!-- <module>core-java-streams-7</module> requires JDK 24-->
         <module>core-java-streams-collect</module>
         <module>core-java-streams-maps</module>
         <module>core-java-string-operations-3</module>

pom.xml

@@ -951,6 +951,7 @@
             </build>
 
             <modules>
+                <module>core-java-modules/core-java-streams-7</module>
                 <module>core-java-modules/core-java-24</module>
             </modules>
 
@@ -1367,6 +1368,7 @@
             </build>
 
             <modules>
+                <module>core-java-modules/core-java-streams-7</module>
                 <module>core-java-modules/core-java-24</module>
             </modules>