Implement exactSize(n) (#40)

+ Going with this implementation for now, not sure if there is much call for a predicate based version of this

CHANGELOG.md

@@ -1,7 +1,8 @@
 ### 0.5.0
 + Implement `reverse()`
 + Implement `maxBy(fn)` and `minBy(fn)`
-
++ Implement `exactSize(n)`
++ 
 ### 0.4.0
 + Implement `suffle()` and `shuffle(RandomGenerator)`
 + Implement `filterWithIndex()`

README.md

@@ -37,6 +37,7 @@ implementation("com.ginsberg:gatherers4j:0.5.0")
 | `dedupeConsecutive()`        | Remove consecutive duplicates from a stream                                                                                    |
 | `dedupeConsecutiveBy(fn)`    | Remove consecutive duplicates from a stream as returned by `fn`                                                                |
 | `distinctBy(fn)`             | Emit only distinct elements from the stream, as measured by `fn`                                                               |
+| `exactSize(n)`               | Ensure the stream is exactly `n` elements long, or throw an `IllegalStateException`                                            |
 | `filterWithIndex(predicate)` | Filter the stream with the given `predicate`, which takes an `element` and its `index`                                         | 
 | `interleave(stream)`         | Creates a stream of alternating objects from the input stream and the argument stream                                          |
 | `last(n)`                    | Constrain the stream to the last `n` values                                                                                    |
@@ -138,6 +139,19 @@ Stream
 // [Person("Todd", "Ginsberg"), Person("Emma", "Ginsberg")]
 ```
 
+#### Ensure the stream is exactly `n` elements long
+
+```java
+// Good
+
+Stream.of("A", "B", "C").gather(Gatherers4j.exactSize(3)).toList();
+// ["A", "B", "C"]
+
+// Bad
+Stream.of("A").gather(Gatherers4j.exactSize(3)).toList();
+// IllegalStateException
+```
+
 #### Filter a stream, knowing the index of each element
 
 ```java

src/main/java/com/ginsberg/gatherers4j/Gatherers4j.java

@@ -75,6 +75,16 @@ public static <INPUT> ThrottlingGatherer<INPUT> debounce(final int amount, final
         return new DistinctGatherer<>(function);
     }
 
+    /**
+     * Ensure the input stream is exactly <code>size</code> elements long, and emit all elements
+     * if so. If not, throw an <code>IllegalStateException</code>.
+     *
+     * @param size Exact number of elements the stream must have
+     */
+    public static <INPUT> SizeGatherer<INPUT> exactSize(final long size) {
+        return new SizeGatherer<>(size);
+    }
+
     /**
      * Filter a stream according to the given <code>predicate</code>, which takes both the item being examined, and its index.
      *
@@ -138,7 +148,7 @@ public static <INPUT, MAPPED extends Comparable<MAPPED>> MinMaxGatherer<INPUT, M
 
     /**
      * Reverse the order of the input Stream.
-     *
+     * <p>
      * Note: This consumes the entire stream and holds it in memory, so it will not work on
      * infinite streams and may cause memory pressure on very large streams.
      */
@@ -243,7 +253,7 @@ public static BigDecimalSimpleAverageGatherer<BigDecimal> simpleRunningAverage()
 
     /**
      * Shuffle the input stream into a random order.
-     *
+     * <p>
      * Note: This consumes the entire stream and holds it in memory, so it will not work on
      * infinite streams and may cause memory pressure on very large streams.
      *

src/main/java/com/ginsberg/gatherers4j/SizeGatherer.java

@@ -0,0 +1,70 @@
+/*
+ * Copyright 2024 Todd Ginsberg
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.ginsberg.gatherers4j;
+
+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 SizeGatherer<INPUT> implements Gatherer<INPUT, SizeGatherer.State<INPUT>, INPUT> {
+
+    private final long targetSize;
+
+    SizeGatherer(long targetSize) {
+        if (targetSize < 0) {
+            throw new IllegalArgumentException("Target size cannot be negative");
+        }
+        this.targetSize = targetSize;
+    }
+
+    @Override
+    public Supplier<State<INPUT>> initializer() {
+        return State::new;
+    }
+
+    @Override
+    public Integrator<State<INPUT>, INPUT, INPUT> integrator() {
+        return (state, element, downstream) -> {
+            state.elements.add(element);
+            if (state.elements.size() > targetSize) {
+                fail();
+            }
+            return !downstream.isRejecting();
+        };
+    }
+
+    @Override
+    public BiConsumer<State<INPUT>, Downstream<? super INPUT>> finisher() {
+        return (state, downstream) -> {
+            if (state.elements.size() == targetSize) {
+                state.elements.forEach(downstream::push);
+            } else {
+                fail();
+            }
+        };
+    }
+
+    private void fail() {
+        throw new IllegalStateException("Size must be exactly " + targetSize);
+    }
+
+    public static class State<INPUT> {
+        final List<INPUT> elements = new ArrayList<>();
+    }
+}

src/test/java/com/ginsberg/gatherers4j/SizeGathererTest.java

@@ -0,0 +1,62 @@
+/*
+ * Copyright 2024 Todd Ginsberg
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.ginsberg.gatherers4j;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.stream.Stream;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+class SizeGathererTest {
+
+    @Test
+    void sizeMustNotBeNegative() {
+        assertThatThrownBy(() ->
+                Stream.empty().gather(Gatherers4j.exactSize(-1)).toList()
+        ).isInstanceOf(IllegalArgumentException.class);
+    }
+
+    @Test
+    void doesNotEmitUnderTarget() {
+        assertThatThrownBy(() ->
+                Stream.of("A").gather(Gatherers4j.exactSize(2)).toList()
+        ).isInstanceOf(IllegalStateException.class);
+    }
+
+    @Test
+    void doesNotEmitOverTarget() {
+        assertThatThrownBy(() ->
+                Stream.of("A", "B", "C").gather(Gatherers4j.exactSize(2)).toList()
+        ).isInstanceOf(IllegalStateException.class);
+    }
+
+    @Test
+    void emitsAtTarget() {
+        // Arrange
+        final Stream<String> input = Stream.of("A", "B", "C");
+
+        // Act
+        final List<String> output = input.gather(Gatherers4j.exactSize(3)).toList();
+
+        // Assert
+        assertThat(output).containsExactly("A", "B", "C");
+    }
+
+}