/*
 * 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 com.ginsberg.gatherers4j.dto.Pair;
import com.ginsberg.gatherers4j.dto.WithCount;
import com.ginsberg.gatherers4j.dto.WithIndex;
import com.ginsberg.gatherers4j.enums.Frequency;
import com.ginsberg.gatherers4j.enums.Order;
import com.ginsberg.gatherers4j.enums.Rotate;
import com.ginsberg.gatherers4j.enums.Size;
import org.jspecify.annotations.Nullable;

import java.math.BigDecimal;
import java.time.Duration;
import java.util.Comparator;
import java.util.Iterator;
import java.util.List;
import java.util.function.BiConsumer;
import java.util.function.BiFunction;
import java.util.function.BiPredicate;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.function.Supplier;
import java.util.random.RandomGenerator;
import java.util.stream.Gatherer;
import java.util.stream.Stream;

import static com.ginsberg.gatherers4j.util.GathererUtils.equalityOnlyComparator;
import static com.ginsberg.gatherers4j.util.GathererUtils.mustNotBeNull;

/// This is the main entry-point for the Gatherers4j library. All available gatherers
/// are created from static methods on this class.
public abstract class Gatherers4j {

    private Gatherers4j() {
        // No
    }

    /// Cross every element of the input stream with every element of the given `Iterable`, emitting them
    /// to the output stream as a `Pair<INPUT, CROSS>`.
    ///
    /// @param <INPUT> Type of element in the input stream
    /// @param <CROSS> Type of element in the crossWith `Iterable`
    /// @param source The Iterable to source with
    /// @return A non-null Gatherer
    public static <INPUT extends @Nullable Object, CROSS extends @Nullable Object> Gatherer<INPUT, ?, Pair<INPUT, CROSS>> crossWith(
            final Iterable<CROSS> source
    ) {
        return CrossGatherer.of(source);
    }

    /// Cross every element of the input stream with every element of the given `Iterator`, emitting them
    /// to the output stream as a `Pair<INPUT, CROSS>`.
    ///
    /// Note: the Iterator is consumed fully and stored as a List in memory.
    ///
    /// @param <INPUT> Type of element in the input stream
    /// @param <CROSS> Type of element in the crossWith `Iterator`
    /// @param source The Iterator to cross with
    /// @return A non-null Gatherer
    public static <INPUT extends @Nullable Object, CROSS extends @Nullable Object> Gatherer<INPUT, ?, Pair<INPUT, CROSS>> crossWith(
            final Iterator<CROSS> source
    ) {
        return CrossGatherer.of(source);
    }

    /// Cross every element of the input stream with every element of the given `Stream`, emitting them
    /// to the output stream as a `Pair<INPUT, CROSS>`.
    ///
    /// Note: the Iterator is consumed fully and stored as a List in memory.
    /// Note: The Ghostbusters warned us about this and I hereby absolve myself of any responsibility if you cause some kind of cataclysm.
    ///
    /// @param <INPUT> Type of element in the input stream
    /// @param <CROSS> Type of element in the crossWith `Iterator`
    /// @param source The Stream to cross with
    /// @return A non-null Gatherer
    public static <INPUT extends @Nullable Object, CROSS extends @Nullable Object> Gatherer<INPUT, ?, Pair<INPUT, CROSS>> crossWith(
            final Stream<CROSS> source
    ) {
        return CrossGatherer.of(source);
    }

    /// Cross every element of the input stream with every element provided, emitting them
    /// to the output stream as a `Pair<INPUT, CROSS>`.
    ///
    /// Note: the Iterator is consumed fully and stored as a List in memory.
    /// Note: The Ghostbusters warned us about this and I hereby absolve myself of any responsibility if you cause some kind of cataclysm.
    ///
    /// @param <INPUT> Type of element in the input stream
    /// @param <CROSS> Type of element in the crossWith `Iterator`
    /// @param source Elements to cross with the input stream
    /// @return A non-null Gatherer
    @SafeVarargs
    public static <INPUT extends @Nullable Object, CROSS extends @Nullable Object> Gatherer<INPUT, ?, Pair<INPUT, CROSS>> crossWith(
            final CROSS... source
    ) {
        return CrossGatherer.of(source);
    }

    /// Limit the number of elements in the stream to some number per period, dropping anything over the
    /// limit during the period.
    ///
    /// @param amount   A positive number of elements to allow per period
    /// @param duration A positive duration for the length of the period
    /// @param <INPUT>  Type of elements in both the input and output streams
    /// @return A non-null `ThrottlingGatherer`
    public static <INPUT extends @Nullable Object> ThrottlingGatherer<INPUT> debounce(
            final int amount,
            final Duration duration
    ) {
        return new ThrottlingGatherer<>(ThrottlingGatherer.LimitRule.Drop, amount, duration);
    }

    /// Remove consecutive duplicate elements from a stream as measured by `Object.equals(Object)`
    ///
    /// @param <INPUT> Type of elements in both the input and output streams
    /// @return A non-null `Gatherer`
    public static <INPUT extends @Nullable Object> Gatherer<INPUT, ?, INPUT> dedupeConsecutive() {
        return new DedupeConsecutiveGatherer<>();
    }

    /// Remove consecutive duplicates from a stream where duplication is measured by the given `function`.
    ///
    /// @param mappingFunction A non-null function, the results of which will be used to check for consecutive duplication.
    /// @param <INPUT>         Type of elements in both the input and output streams
    /// @return A non-null `Gatherer`
    public static <INPUT extends @Nullable Object> Gatherer<INPUT, ?, INPUT> dedupeConsecutiveBy(
            final Function<INPUT, Object> mappingFunction
    ) {
        mustNotBeNull(mappingFunction, "Mapping function must not be null");
        return new DedupeConsecutiveGatherer<>(mappingFunction);
    }

    /// Filter a stream such that it only contains distinct elements measured by the given `function`.
    ///
    /// @param mappingFunction A non-null mapping function, the results of which will be used to check for distinct elements
    /// @param <INPUT>         Type of elements in both the input and output streams
    /// @return A non-null `Gatherer`
    public static <INPUT extends @Nullable Object> Gatherer<INPUT, ?, INPUT> distinctBy(
            final Function<INPUT, @Nullable Object> mappingFunction
    ) {
        return new DistinctGatherer<>(mappingFunction);
    }

    /// Drop every nth element of the stream.
    ///
    /// @param count   The number of the elements to drop, must be at least 2
    /// @param <INPUT> Type of elements in both the input and output streams
    /// @return A non-null `Gatherer`
    public static <INPUT extends @Nullable Object> Gatherer<INPUT, ?, INPUT> dropEveryNth(final int count) {
        if (count < 2) {
            throw new IllegalArgumentException("Count must be a minimum of 2");
        }
        return filterIndexed((index, _) -> index % count != 0);
    }

    /// Keep all elements except the last `count` elements of the stream.
    ///
    /// @param count   A positive number of elements to drop from the end of the stream
    /// @param <INPUT> Type of elements in both the input and output streams
    /// @return A non-null `Gatherer`
    public static <INPUT extends @Nullable Object> Gatherer<INPUT, ?, INPUT> dropLast(final int count) {
        return new DropLastGatherer<>(count);
    }

    /// Ensure that the `Comparable` elements in the input stream are in the given `Order`, and fail exceptionally if they are not.
    ///
    /// @param <INPUT> Type of elements in the input stream
    /// @param order The non-null order the stream must be in.
    /// @return A non-null Gatherer
    public static <INPUT extends Comparable<INPUT>> Gatherer<INPUT, ?, INPUT> ensureOrdered(final Order order) {
        final Gatherer<INPUT, ?, List<INPUT>> generic = GroupChangingGatherer.usingComparable(order);
        return generic.andThen(new FlattenSingleOrFail<>("Elements not in proper order: " + order.name()));
    }

    /// Ensure that the elements in the input stream are in the given `Order` as measured by the given `Comparator`, and fail exceptionally if they are not.
    ///
    /// @param <INPUT> Type of elements in the input stream
    /// @param order The non-null order the stream must be in.
    /// @param comparator The non-null comparator used to compare stream elements
    /// @return A non-null Gatherer
    public static <INPUT> Gatherer<INPUT, ?, INPUT> ensureOrderedBy(final Order order, final Comparator<INPUT> comparator) {
        return GroupChangingGatherer.usingComparator(order, comparator)
                .andThen(new FlattenSingleOrFail<>("Elements not in proper order: " + order.name()));
    }

    /// Ensure the input stream's meets the given `size` criteria, and emit all elements if so.
    /// If not, throw an `IllegalStateException`.
    ///
    /// @param size The Size to measure the stream length against
    /// @param length    Number to compare stream length against
    /// @param <INPUT> Type of elements in both the input and output streams
    /// @return A non-null `SizeGatherer`
    /// @throws IllegalStateException when the input stream is not exactly `size` elements long
    public static <INPUT extends @Nullable Object> SizeGatherer<INPUT> ensureSize(final Size size, final long length) {
        return new SizeGatherer<>(size, length);
    }

    /// Create a Stream that represents the exponential moving average of a `Stream<BigDecimal>`, with the given `alpha`.
    ///
    /// @param alpha The alpha value to use in the EMA calculation.
    /// @return A non-null `BigDecimalExponentialMovingAverageGatherer`
    public static BigDecimalExponentialMovingAverageGatherer<@Nullable BigDecimal> exponentialMovingAverageWithAlpha(final double alpha) {
        return BigDecimalExponentialMovingAverageGatherer.withAlpha(alpha, Function.identity());
    }

    /// Create a Stream that represents the exponential moving average of `BigDecimal` objects mapped from a `Stream<INPUT>`
    /// via a `mappingFunction` and using the given `alpha`.
    ///
    /// @param alpha The alpha value to use in the EMA calculation.
    /// @param mappingFunction A function to map `<INPUT>` objects to `BigDecimal`, the results of which will be used
    ///                        in the exponential average calculation
    /// @param <INPUT>         Type of elements in the input stream, to be remapped to `BigDecimal` by the `mappingFunction`
    /// @return A non-null `BigDecimalExponentialMovingAverageGatherer`
    public static <INPUT extends @Nullable Object> BigDecimalExponentialMovingAverageGatherer<INPUT> exponentialMovingAverageWithAlphaBy(
            final double alpha,
            final Function<INPUT, BigDecimal> mappingFunction
    ) {
        return BigDecimalExponentialMovingAverageGatherer.withAlpha(alpha, mappingFunction);
    }

    /// Create a Stream that represents the exponential moving average of a `Stream<BigDecimal>`, over the given number of `periods`.
    ///
    /// @param periods The number of periods to use in the EMA calculation.
    /// @return A non-null `BigDecimalExponentialMovingAverageGatherer`
    public static BigDecimalExponentialMovingAverageGatherer<@Nullable BigDecimal> exponentialMovingAverageWithPeriod(final int periods) {
        return BigDecimalExponentialMovingAverageGatherer.withPeriod(periods, Function.identity());
    }

    /// Create a Stream that represents the exponential moving average of `BigDecimal` objects mapped from a `Stream<INPUT>`
    /// via a `mappingFunction` over the given number of `periods`.
    ///
    /// @param periods The number of periods to use in the EMA calculation.
    /// @param mappingFunction A function to map `<INPUT>` objects to `BigDecimal`, the results of which will be used
    ///                        in the exponential average calculation
    /// @param <INPUT>         Type of elements in the input stream, to be remapped to `BigDecimal` by the `mappingFunction`
    /// @return A non-null `BigDecimalExponentialMovingAverageGatherer`
    public static <INPUT extends @Nullable Object> BigDecimalExponentialMovingAverageGatherer<INPUT> exponentialMovingAverageWithPeriodBy(
            final int periods,
            final Function<INPUT, BigDecimal> mappingFunction
    ) {
        return BigDecimalExponentialMovingAverageGatherer.withPeriod(periods, mappingFunction);
    }

    /// Filter a stream according to the given `predicate`, which takes both the item being examined,
    /// and its index.
    ///
    /// @param predicate A non-null `BiPredicate<Integer,INPUT>` where the `Integer` is the zero-based index of the element
    ///                  being filtered, and the `INPUT` is the element itself.
    /// @param <INPUT>   Type of elements in the input stream
    /// @return A non-null `Gatherer`
    public static <INPUT extends @Nullable Object> Gatherer<INPUT, ?, INPUT> filterIndexed(
            final BiPredicate<Integer, INPUT> predicate
    ) {
        return SimpleIndexingGatherers.filterIndexed(predicate);
    }

    /// Filter the elements in the stream to only include elements of the given types.
    /// Note, due to how generics work you may end up with some... interesting stream types as a result
    ///
    /// @param <INPUT> Type of elements in the input stream
    /// @param <OUTPUT> Type of elements in the output stream
    /// @param validTypes A non-empty array of types to filter for
    /// @return A non-null `Gatherer`
    @SafeVarargs
    public static <INPUT extends @Nullable Object, OUTPUT extends @Nullable Object> Gatherer<INPUT, ?, OUTPUT> filterInstanceOf(
            final Class<? extends OUTPUT>... validTypes
    ) {
        return TypeFilteringGatherer.of(validTypes);
    }

    /// Filter the input stream so that it contains `Comparable` elements in the `order` specified. Anything not matching
    /// that order is removed as it is encountered.
    ///
    /// @param <INPUT> Type of elements in the input and output stream
    /// @return A non-null gatherer
    public static <INPUT extends Comparable<INPUT>> Gatherer<INPUT, ?, INPUT> filterOrdered(final Order order) {
        return FilterChangingGatherer.usingComparable(order);
    }

    /// Filter the input stream so that it contains elements in the `order` specified as measured by the given `Comparator`.
    /// Anything not matching that order is removed as it is encountered.
    ///
    /// @param <INPUT> Type of elements in the input and output stream
    /// @param comparator A non-null `Comparator` to compare stream elements
    /// @return A non-null gatherer
    public static <INPUT> Gatherer<INPUT, ?, INPUT> filterOrderedBy(final Order order, final Comparator<INPUT> comparator) {
        return FilterChangingGatherer.usingComparator(order, comparator);
    }

    ///  Perform a fold over every element in the input stream along with its index
    ///
    /// @param <INPUT>      Type of elements in the input stream
    /// @param <OUTPUT>     Type elements are folded to (the accumulated value)
    /// @param initialValue Initial value of the fold
    /// @param foldFunction Function that performs the fold given an element, its index, and the carry value
    /// @return A non-null Gatherer
    public static <INPUT extends @Nullable Object, OUTPUT extends @Nullable Object> Gatherer<INPUT, ?, OUTPUT> foldIndexed(
            final Supplier<OUTPUT> initialValue,
            final IndexedAccumulatorFunction<? super OUTPUT, ? super INPUT, ? extends OUTPUT> foldFunction
    ) {
        return new AccumulatingGatherer<>(false, initialValue, foldFunction);
    }

    /// Turn a `Stream<INPUT>` into a `Stream<List<INPUT>>` where adjacent equal elements are in the same `List`
    /// and equality is measured by `Object.equals(Object)`. The lists emitted to the output stream are unmodifiable.
    ///
    /// @param <INPUT> Type of elements in the input stream
    /// @return A non-null `GroupingByGatherer`
    public static <INPUT extends @Nullable Object> Gatherer<INPUT, ?, List<INPUT>> group() {
        return GroupChangingGatherer.usingComparator(Order.Equal, equalityOnlyComparator());
    }

    /// Turn a `Stream<INPUT>` into a `Stream<List<INPUT>>` where adjacent equal elements are in the same `List`
    /// and equality is measured by the given `mappingFunction`. The lists emitted to the output stream are unmodifiable.
    ///
    /// @param mappingFunction A non-null function, the results of which are used to measure equality of consecutive elements.
    /// @param <INPUT>         Type of elements in the input stream
    /// @return A non-null `GroupingByGatherer`
    public static <INPUT extends @Nullable Object> Gatherer<INPUT, ?, List<INPUT>> groupBy(
            final Function<@Nullable INPUT, @Nullable Object> mappingFunction
    ) {
        mustNotBeNull(mappingFunction, "mappingFunction must not be null");
        return GroupChangingGatherer.usingComparator(Order.Equal, equalityOnlyComparator(mappingFunction));
    }

    /// Turn a `Stream<Comparable>` into a `Stream<List<>>` where adjacent equal elements are in the same `List`
    /// and order is measured by the order imposed by the `Comparable`. The lists emitted to the output stream are unmodifiable.
    ///
    /// @param <INPUT> Type of elements in the input stream, implementing `Comparable`
    /// @return A non-null `Gatherer`
    public static <INPUT extends @Nullable Comparable<INPUT>> Gatherer<INPUT, ?, List<INPUT>> groupOrdered(final Order order) {
        return GroupChangingGatherer.usingComparable(order);
    }

    /// Turn a `Stream<INPUT>` into a `Stream<List<INPUT>>` where adjacent equal elements are in the same `List`
    /// and order is measured by the given `Comparator`. The lists emitted to the output stream are unmodifiable.
    ///
    /// @param comparator A non-null function, the results of which are used to measure equality of consecutive elements.
    /// @param <INPUT>         Type of elements in the input stream
    /// @return A non-null `Gatherer`
    public static <INPUT extends @Nullable Object> Gatherer<INPUT, ?, List<INPUT>> groupOrderedBy(
            final Order order,
            final Comparator<INPUT> comparator
    ) {
        return GroupChangingGatherer.usingComparator(order, comparator);
    }

    /// Creates a stream of alternating objects from the input stream and the argument iterable
    ///
    /// @param other   A non-null Iterable to interleave
    /// @param <INPUT> Type of elements in both the input stream and argument iterable
    /// @return A non-null `InterleavingGatherer`
    public static <INPUT extends @Nullable Object> InterleavingGatherer<INPUT> interleaveWith(final Iterable<INPUT> other) {
        return new InterleavingGatherer<>(other);
    }

    /// Creates a stream of alternating objects from the input stream and the argument iterator
    ///
    /// @param other   A non-null Iterator to interleave
    /// @param <INPUT> Type of elements in both the input stream and argument iterator
    /// @return A non-null `InterleavingGatherer`
    public static <INPUT extends @Nullable Object> InterleavingGatherer<INPUT> interleaveWith(final Iterator<INPUT> other) {
        return new InterleavingGatherer<>(other);
    }

    /// Creates a stream of alternating objects from the input stream and the argument stream
    ///
    /// @param other   A non-null stream to interleave
    /// @param <INPUT> Type of elements in both the input and argument streams
    /// @return A non-null `InterleavingGatherer`
    public static <INPUT extends @Nullable Object> InterleavingGatherer<INPUT> interleaveWith(final Stream<INPUT> other) {
        return new InterleavingGatherer<>(other);
    }

    /// Creates a stream of alternating objects from the input stream and the provided elements
    ///
    /// @param other   Non-null elements to interleave
    /// @param <INPUT> Type of elements in both the input stream and argument iterator
    /// @return A non-null `InterleavingGatherer`
    @SafeVarargs
    public static <INPUT extends @Nullable Object> InterleavingGatherer<INPUT> interleaveWith(final INPUT... other) {
        return new InterleavingGatherer<>(other);
    }

    /// Intersperse the given `intersperseElement` between each element of the input stream.
    ///
    /// @param intersperseElement The element to intersperse, which may be null
    /// @param <INPUT> The type of elements in the stream and the element to intersperse
    /// @return A non-null Gatherer
    public static <INPUT extends @Nullable Object> Gatherer<INPUT, ?, INPUT> intersperse(final INPUT intersperseElement) {
        return new IntersperseGatherer<>(intersperseElement);
    }

    /// Perform a mapping operation given the element being mapped and its zero-based index.
    ///
    /// @param <INPUT> The type of elements in the input stream
    /// @param <OUTPUT> The type of elements in the output stream
    /// @param mappingFunction A non-null function to map input to output, given an input and its index
    /// @return A non-null Gatherer
    public static <INPUT extends @Nullable Object, OUTPUT extends @Nullable Object> Gatherer<INPUT, ?, OUTPUT> mapIndexed(
            final BiFunction<Integer, INPUT, OUTPUT> mappingFunction) {
        return SimpleIndexingGatherers.mapIndexed(mappingFunction);
    }

    /// Create a stream that represents the moving maximum value over the previous `windowSize` elements.
    ///
    /// @param <INPUT> Type of elements in the input and output stream; must implement `Comparable`
    /// @param windowSize How large the lookback window is, must be at least 2
    /// @return A non-null `MinMaxGatherer`
    public static <INPUT extends @Nullable Comparable<INPUT>> MinMaxGatherer<INPUT> movingMax(final int windowSize) {
        return MinMaxGatherer.movingUsingComparable(windowSize, false);
    }

    /// Create a stream that represents the moving maximum value over the previous `windowSize` elements,
    /// as calculated by the given `Comparator`.
    ///
    /// @param <INPUT> Type of elements in the input and output stream
    /// @param windowSize How large the lookback window is, must be at least 2
    /// @param comparator A non-null comparator to use to compare stream elements
    /// @return A non-null `MinMaxGatherer`
    public static <INPUT extends @Nullable Object> MinMaxGatherer<INPUT> movingMaxBy(
            final int windowSize,
            final Comparator<INPUT> comparator
    ) {
        return MinMaxGatherer.movingUsingComparator(windowSize, false, comparator);
    }

    /// Create a Stream that represents the moving median of a `Stream<BigDecimal>` looking
    /// back `windowSize` number of elements.
    ///
    /// @param windowSize The trailing number of elements to calculate the median over, must be greater than 1.
    /// @return A non-null `BigDecimalMovingProductGatherer`
    public static BigDecimalMedianGatherer<@Nullable BigDecimal> movingMedian(final int windowSize) {
        if (windowSize <= 1) {
            throw new IllegalArgumentException("windowSize must be greater than 1");
        }
        return new BigDecimalMedianGatherer<>(windowSize, Function.identity());
    }

    /// Create a Stream that represents the moving median of `BigDecimal` objects mapped from a `Stream<INPUT>`
    /// via a `mappingFunction` and looking back `windowSize` number of elements.
    ///
    /// @param windowSize      The trailing number of elements to calculate the median from. Must be greater than 1.
    /// @param mappingFunction A function to map `<INPUT>` objects to `BigDecimal`, the results of which will be used
    ///                        in the moving product calculation
    /// @param <INPUT>         Type of elements in the input stream, to be remapped to `BigDecimal` by the `mappingFunction`
    /// @return A non-null `BigDecimalMovingProductGatherer`
    public static <INPUT extends @Nullable Object> BigDecimalMedianGatherer<INPUT> movingMedianBy(
            final int windowSize,
            final Function<INPUT, BigDecimal> mappingFunction
    ) {
        if (windowSize <= 1) {
            throw new IllegalArgumentException("windowSize must be greater than 1");
        }
        return new BigDecimalMedianGatherer<>(windowSize, mappingFunction);
    }

    /// Create a stream that represents the moving minimum value over the previous `windowSize` elements.
    ///
    /// @param <INPUT> Type of elements in the input and output stream; must implement `Comparable`
    /// @param windowSize How large the lookback window is, must be at least 2
    /// @return A non-null `MinMaxGatherer`
    public static <INPUT extends @Nullable Comparable<INPUT>> MinMaxGatherer<INPUT> movingMin(final int windowSize) {
        return MinMaxGatherer.movingUsingComparable(windowSize, true);
    }

    /// Create a stream that represents the moving minimum value over the previous `windowSize` elements,
    /// as calculated by the given `Comparator`.
    ///
    /// @param <INPUT> Type of elements in the input and output stream
    /// @param windowSize How large the lookback window is, must be at least 2
    /// @param comparator A non-null comparator to use to compare stream elements
    /// @return A non-null `MinMaxGatherer`
    public static <INPUT extends @Nullable Object> MinMaxGatherer<INPUT> movingMinBy(
            final int windowSize,
            final Comparator<INPUT> comparator
    ) {
        return MinMaxGatherer.movingUsingComparator(windowSize, true, comparator);
    }

    /// Create a Stream that represents the moving product of a `Stream<BigDecimal>` looking
    /// back `windowSize` number of elements.
    ///
    /// @param windowSize The trailing number of elements to multiply, must be greater than 1.
    /// @return A non-null `BigDecimalMovingProductGatherer`
    public static BigDecimalMovingProductGatherer<@Nullable BigDecimal> movingProduct(final int windowSize) {
        return new BigDecimalMovingProductGatherer<>(windowSize, Function.identity());
    }

    /// Create a Stream that represents the moving product of `BigDecimal` objects mapped from a `Stream<INPUT>`
    /// via a `mappingFunction` and looking back `windowSize` number of elements.
    ///
    /// @param windowSize      The trailing number of elements to multiply, must be greater than 1.
    /// @param mappingFunction A function to map `<INPUT>` objects to `BigDecimal`, the results of which will be used
    ///                        in the moving product calculation
    /// @param <INPUT>         Type of elements in the input stream, to be remapped to `BigDecimal` by the `mappingFunction`
    /// @return A non-null `BigDecimalMovingProductGatherer`
    public static <INPUT extends @Nullable Object> BigDecimalMovingProductGatherer<INPUT> movingProductBy(
            final int windowSize,
            final Function<INPUT, BigDecimal> mappingFunction
    ) {
        return new BigDecimalMovingProductGatherer<>(windowSize, mappingFunction);
    }

    /// Create a Stream that represents the moving sum of a `Stream<BigDecimal>` looking
    /// back `windowSize` number of elements.
    ///
    /// @param windowSize The trailing number of elements to add, must be greater than 1.
    /// @return A non-null `BigDecimalMovingSumGatherer`
    public static BigDecimalMovingSumGatherer<@Nullable BigDecimal> movingSum(final int windowSize) {
        return new BigDecimalMovingSumGatherer<>(windowSize, Function.identity());
    }

    /// Create a Stream that represents the moving sum of `BigDecimal` objects mapped from a `Stream<INPUT>`
    /// via a `mappingFunction` and looking back `windowSize` number of elements.
    ///
    /// @param windowSize      The trailing number of elements to multiply, must be greater than 1.
    /// @param mappingFunction A function to map `<INPUT>` objects to `BigDecimal`, the results of which will be used
    ///                        in the moving sum calculation
    /// @param <INPUT>         Type of elements in the input stream, to be remapped to `BigDecimal` by the `mappingFunction`
    /// @return A non-null `BigDecimalMovingSumGatherer`
    public static <INPUT extends @Nullable Object> BigDecimalMovingSumGatherer<INPUT> movingSumBy(
            final int windowSize,
            final Function<INPUT, BigDecimal> mappingFunction
    ) {
        return new BigDecimalMovingSumGatherer<>(windowSize, mappingFunction);
    }

    /// Emit elements in the input stream ordered by frequency in the direction specified. Elements are emitted wrapped
    /// in `WithCount<INPUT>` objects that carry the element and the number of occurrences.
    ///
    /// 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.
    ///
    /// @param <INPUT> Type of elements in the input stream
    /// @return A non-null `Gatherer`
    public static <INPUT extends @Nullable Object> Gatherer<INPUT, ?, WithCount<INPUT>> orderByFrequency(final Frequency order) {
        return new FrequencyGatherer<>(order);
    }

    /// Peek at each element along with its zero-based index.
    ///
    /// @param <INPUT> The type of elements in the input stream
    /// @param peekingConsumer A non-null consumer to peek at each element and its index
    /// @return A non-null Gatherer
    public static <INPUT extends @Nullable Object> Gatherer<INPUT, ?, INPUT> peekIndexed(
            final BiConsumer<Integer, INPUT> peekingConsumer) {
        return SimpleIndexingGatherers.peekIndexed(peekingConsumer);
    }

    /// Repeatedly emit the input stream to the output stream a given number of times.
    /// Note: This implementation consumes the entire input stream into memory, so it must be used on finite streams.
    ///
    /// @param <INPUT> Type of elements in the input and output stream
    /// @param repeats Number of repeats, must be greater than 1
    /// @return A non-null `Gatherer`
    public static <INPUT extends @Nullable Object> Gatherer<INPUT, ?, INPUT> repeat(final int repeats) {
        return RepeatingGatherer.ofFinite(repeats);
    }

    /// Repeatedly emit the input stream to the output stream infinitely.
    /// Note: This implementation consumes the entire input stream into memory, so it must be used on finite streams.
    ///
    /// @param <INPUT> Type of elements in the input and output stream
    /// @return A non-null `Gatherer`
    public static <INPUT extends @Nullable Object> Gatherer<INPUT, ?, INPUT> repeatInfinitely() {
        return RepeatingGatherer.ofInfinite();
    }

    /// Reverse the order of the input Stream.
    ///
    /// 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.
    ///
    /// @param <INPUT> Type of elements in the input stream
    /// @return A non-null `Gatherer`
    public static <INPUT extends @Nullable Object> Gatherer<INPUT, ?, INPUT> reverse() {
        return new ReversingGatherer<>();
    }

    /// Consume the entire stream and emit its elements rotated in the direction specified `distance` number of spaces
    ///
    /// @param <INPUT> Type of elements in the input and output stream
    /// @param direction Which direction to rotate the stream in
    /// @param distance Distance to rotate elements
    /// @return A non-null Gatherer
    public static <INPUT extends @Nullable Object> Gatherer<INPUT, ?, INPUT> rotate(final Rotate direction, final int distance) {
        return new RotateGatherer<>(direction, distance);
    }

    /// Emit the running maximum value of `Comparable` elements in the input stream.
    ///
    /// @param <INPUT> Type of elements in the input and output stream; must implement `Comparable`
    /// @return A non-null `MinMaxGatherer`
