Implement filterWithIndex() (#31)

+ Remove `withIndexStartingAt` - does not seem useful

CHANGELOG.md

@@ -1,6 +1,8 @@
 ### 0.4.0
-+ Remove `concat` implementation (the JDK has this)
 + Implement `suffle()` and `shuffle(RandomGenerator)`
++ Implement `filterWithIndex()`
++ Remove `concat()` implementation (the JDK has this)
++ Remove `withIndexStartingAt()`, not sure if this is a real use case.
 
 ### 0.3.0
 + Move minimum Java version from 22 to 23

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

@@ -0,0 +1,53 @@
+/*
+ * 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.function.BiPredicate;
+import java.util.function.Supplier;
+import java.util.stream.Gatherer;
+
+import static com.ginsberg.gatherers4j.GathererUtils.mustNotBeNull;
+
+public class FilteringWithIndexGatherer<INPUT>
+        implements Gatherer<INPUT, FilteringWithIndexGatherer.State, INPUT> {
+
+    private final BiPredicate<Long, INPUT> predicate;
+
+    FilteringWithIndexGatherer(final BiPredicate<Long, INPUT> predicate) {
+        mustNotBeNull(predicate, "Predicate must not be null");
+        this.predicate = predicate;
+    }
+
+    @Override
+    public Supplier<FilteringWithIndexGatherer.State> initializer() {
+        return State::new;
+    }
+
+    @Override
+    public Integrator<FilteringWithIndexGatherer.State, INPUT, INPUT> integrator() {
+        return (state, element, downstream) -> {
+            if (predicate.test(state.index++, element)) {
+                downstream.push(element);
+            }
+            return !downstream.isRejecting();
+        };
+    }
+
+    public static class State {
+        long index;
+    }
+}

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

@@ -19,6 +19,7 @@
 import java.math.BigDecimal;
 import java.time.Duration;
 import java.util.List;
+import java.util.function.BiPredicate;
 import java.util.function.Function;
 import java.util.random.RandomGenerator;
 import java.util.stream.Gatherer;
@@ -74,6 +75,12 @@ public static <INPUT> ThrottlingGatherer<INPUT> debounce(final int amount, final
         return new DistinctGatherer<>(function);
     }
 
+    public static <INPUT> FilteringWithIndexGatherer<INPUT> filterWithIndex(
+            final BiPredicate<Long, INPUT> predicate
+    ) {
+        return new FilteringWithIndexGatherer<>(predicate);
+    }
+
     /**
      * Creates a stream of alternating objects from the input stream and the argument stream
      *
@@ -230,15 +237,6 @@ public static <INPUT> IndexingGatherer<INPUT> withIndex() {
         return new IndexingGatherer<>();
     }
 
-    /**
-     * Maps all elements of the stream as-is along with and index, starting at the specified number.
-     *
-     * @param start The starting index to use
-     */
-    public static <INPUT> IndexingGatherer<INPUT> withIndexStartingAt(final long start) {
-        return new IndexingGatherer<INPUT>().startingAt(start);
-    }
-
     /**
      * Creates a stream of `Pair` objects whose values come from the input stream and argument stream
      *

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

@@ -22,19 +22,12 @@
 public class IndexingGatherer<INPUT>
         implements Gatherer<INPUT, IndexingGatherer.State, IndexedValue<INPUT>> {
 
-    private long start = 0;
-
     IndexingGatherer() {
     }
 
-    public IndexingGatherer<INPUT> startingAt(long start) {
-        this.start = start;
-        return this;
-    }
-
     @Override
     public Supplier<IndexingGatherer.State> initializer() {
-        return () -> new State(start);
+        return () -> new State(0);
     }
 
     @Override

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

@@ -0,0 +1,50 @@
+/*
+ * 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 FilteringWithIndexGathererTest {
+
+    @Test
+    void filterWithIndex() {
+        // Arrange
+        final Stream<String> input = Stream.of("A", "B", "C", "D");
+
+        // Act
+        final List<String> output = input
+                .gather(Gatherers4j.filterWithIndex((index, element) ->
+                        index % 2 == 0 || element.equals("D"))
+                )
+                .toList();
+
+        // Assert
+        assertThat(output).containsExactly("A", "C", "D");
+    }
+
+    @Test
+    void predicateMustNotBeNull() {
+        assertThatThrownBy(() -> Stream.of("A").gather(Gatherers4j.filterWithIndex(null)))
+                .isExactlyInstanceOf(IllegalArgumentException.class);
+    }
+}

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

@@ -62,43 +62,4 @@ void integerWithIndex() {
                         new IndexedValue<>(2, 3)
                 );
     }
-
-    @Test
-    void startingAtNonZero() {
-        // Arrange
-        final Stream<String> input = Stream.of("A", "B", "C");
-
-        // Act
-        final List<IndexedValue<String>> output = input
-                .gather(Gatherers4j.withIndexStartingAt(100))
-                .toList();
-
-        // Assert
-        assertThat(output)
-                .containsExactly(
-                        new IndexedValue<>(100, "A"),
-                        new IndexedValue<>(101, "B"),
-                        new IndexedValue<>(102, "C")
-                );
-    }
-
-    @Test
-    void startingAtNonZeroTypeWitnessed() {
-        // Arrange
-        final Stream<String> input = Stream.of("A", "B", "C");
-
-        // Act
-        final List<IndexedValue<String>> output = input
-                .gather(Gatherers4j.<String>withIndex().startingAt(100))
-                .toList();
-
-        // Assert
-        assertThat(output)
-                .containsExactly(
-                        new IndexedValue<>(100, "A"),
-                        new IndexedValue<>(101, "B"),
-                        new IndexedValue<>(102, "C")
-                );
-    }
-
 }