GH-88: Implement Grouping, Filtering, and Valiation of Increasing/Dec…

…reasing/NonIncreasing/NonDecreasing (#93)

+ Group elements to Lists so long as they are increasing, decreasing, non-increasing, or non-decreasing
+ Ensure that streams are increasing, decreasing, non-increasing, or non-decreasing and fail otherwise
+ Filter streams to be increasing, decreasing, non-increasing, or non-decreasing
+ Group, Filter, and Ensure functions can operate on a stream of `Comparable` objects or caller can specify a `Comparator`

CHANGELOG.md

@@ -1,6 +1,9 @@
 ### 0.9.0
 + GH-86: Implement `filterInstanceOf` to filter a stream by type more easily (thanks @nipafx) 
 + Implement `windowed` to provide more options to windowing functions, namely - ability to specify size, how many to skip each time, and whether to include partial windows
++ GH-88: Implement `groupIncreasing`, `groupDecreasing`, `groupNonIncreasing`, and `groupNonDecreasing` with both `Comparable` stream inputs or using an explicit `Comparator` to appropriately group elements in the input stream to lists in the output stream (thanks @nipafx)
++ Implement `ensureIncreasing`, `ensureDecreasing`, `ensureNonIncreasing` and `ensureNonDecreasing` with both `Comparable` stream inputs or using an explicit `Comparator` to ensure the given stream meets the criteria, or fail exceptionally otherwise
++ Implement `filterIncreasing`, `filterDecreasing`, `filterNonIncreasing` and `filterNonDecreasing` with both `Comparable` stream inputs or using an explicit `Comparator` to remove non-compliant elements from the input stream
 
 ### 0.8.0 
 + Add support for `orElse()` and `orElseEmpty()` on size-based gatherers to provide a non-exceptional output stream

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

@@ -0,0 +1,46 @@
+/*
+ * Copyright 2025 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;
+
+enum ChangingOperation {
+    Decreasing {
+        @Override
+        boolean allows(final int comparison) {
+            return comparison < 0;
+        }
+    },
+    Increasing {
+        @Override
+        boolean allows(final int comparison) {
+            return comparison > 0;
+        }
+    },
+    NonDecreasing {
+        @Override
+        boolean allows(final int comparison) {
+            return comparison >= 0;
+        }
+    },
+    NonIncreasing {
+        @Override
+        boolean allows(final int comparison) {
+            return comparison <= 0;
+        }
+    };
+
+    abstract boolean allows(final int comparison);
+}

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

@@ -0,0 +1,85 @@
+/*
+ * Copyright 2025 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.jspecify.annotations.Nullable;
+
+import java.util.Comparator;
+import java.util.function.Supplier;
+import java.util.stream.Gatherer;
+
+import static com.ginsberg.gatherers4j.GathererUtils.mustNotBeNull;
+
+public class FilterChangingGatherer<INPUT>
+        implements Gatherer<INPUT, FilterChangingGatherer.State<INPUT>, INPUT> {
+
+    private final ChangingOperation operation;
+    private final Comparator<INPUT> comparator;
+
+    static <INPUT> FilterChangingGatherer<INPUT> usingComparator(
+            final ChangingOperation operation,
+            final Comparator<INPUT> comparator
+    ) {
+        return new FilterChangingGatherer<>(operation, comparator);
+    }
+
+    static <INPUT extends Comparable<INPUT>> FilterChangingGatherer<INPUT> usingComparable(
+            final ChangingOperation operation
+    ) {
+        return new FilterChangingGatherer<>(operation, Comparable::compareTo);
+    }
+
+    FilterChangingGatherer(
+            final ChangingOperation operation,
+            final Comparator<INPUT> comparator
+    ) {
+        mustNotBeNull(operation, "Operation must not be null");
+        mustNotBeNull(comparator, "Comparator must not be null");
+        this.operation = operation;
+        this.comparator = comparator;
+    }
+
+    @Override
+    public Supplier<FilterChangingGatherer.State<INPUT>> initializer() {
+        return State::new;
+    }
+
+    @Override
+    public Integrator<FilterChangingGatherer.State<INPUT>, INPUT, INPUT> integrator() {
+        return Integrator.ofGreedy((state, element, downstream) -> {
+            if (state.first) {
+                downstream.push(element);
+                state.previousElement = element;
+                state.first = false;
+            } else if (allow(state.previousElement, element)) {
+                downstream.push(element);
+                state.previousElement = element;
+            }
+            return !downstream.isRejecting();
+        });
+    }
+
+    boolean allow(final @Nullable INPUT previous, final INPUT next) {
+        return operation.allows(comparator.compare(next, previous));
+    }
+
+    public static class State<INPUT> {
+        boolean first = true;
+        @Nullable
+        INPUT previousElement;
+    }
+}

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

@@ -0,0 +1,67 @@
+/*
+ * Copyright 2025 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.jspecify.annotations.Nullable;
+
+import java.util.Collection;
+import java.util.function.BiConsumer;
+import java.util.function.Supplier;
+import java.util.stream.Gatherer;
+
+/// Note: "Single" in this case means at most one. The naming of this more precisely seemed clumsy.
+class FlattenSingleOrFail<INPUT extends Collection<OUTPUT>, OUTPUT>
+        implements Gatherer<INPUT, FlattenSingleOrFail.State<INPUT>, OUTPUT> {
+
+    private final String message;
+
+    FlattenSingleOrFail(final String message) {
+        this.message = message;
+    }
+
+    @Override
+    public Supplier<State<INPUT>> initializer() {
+        return State::new;
+    }
+
+    @Override
+    public Integrator<State<INPUT>, INPUT, OUTPUT> integrator() {
+        return (state, element, downstream) -> {
+            if (state.isFirst) {
+                state.firstCollection = element;
+                state.isFirst = false;
+                return !downstream.isRejecting();
+            } else {
+                throw new IllegalStateException(message);
+            }
+        };
+    }
+
+    @Override
+    public BiConsumer<State<INPUT>, Downstream<? super OUTPUT>> finisher() {
+        return (inputState, downstream) -> {
+            if(inputState.firstCollection != null) {
+                inputState.firstCollection.forEach(downstream::push);
+            }
+        };
+    }
+
+    public static class State<INPUT> {
+        boolean isFirst = true;
+        @Nullable INPUT firstCollection;
+    }
+}