/*
 * Copyright 2015, 2024 StreamEx contributors
 *
 * 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 one.util.streamex;

import org.jspecify.annotations.NullMarked;
import org.jspecify.annotations.Nullable;

import java.util.*;
import java.util.Map.Entry;
import java.util.function.*;
import java.util.stream.Collector;
import java.util.stream.Collector.Characteristics;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import static one.util.streamex.AbstractStreamEx.addToMap;
import static one.util.streamex.Internals.*;

/**
 * Implementations of several collectors in addition to ones available in JDK.
 * 
 * @author Tagir Valeev
 * @see Collectors
 * @see Joining
 * @since 0.3.2
 */
@NullMarked
public final class MoreCollectors {
    private MoreCollectors() {
        throw new UnsupportedOperationException();
    }

    /**
     * Returns a {@code Collector} which just ignores the input and calls the
     * provided supplier once to return the output.
     * 
     * @param <T> the type of input elements
     * @param <U> the type of output
     * @param supplier the supplier of the output
     * @return a {@code Collector} which just ignores the input and calls the
     *         provided supplier once to return the output.
     */
    private static <T extends @Nullable Object, U extends @Nullable Object> Collector<T, ?, U> empty(
            Supplier<U> supplier) {
        return new CancellableCollectorImpl<>(() -> NONE, (acc, t) -> {
            // empty
        }, selectFirst(), acc -> supplier.get(), alwaysTrue(), EnumSet.of(Characteristics.UNORDERED,
            Characteristics.CONCURRENT));
    }

    private static <T extends @Nullable Object> Collector<T, ?, List<T>> empty() {
        return empty(ArrayList::new);
    }

    /**
     * Returns a {@code Collector} that accumulates the input elements into a
     * new array.
     *
     * <p>
     * The operation performed by the returned collector is equivalent to
     * {@code stream.toArray(generator)}. This collector is mostly useful as a
     * downstream collector.
     *
     * @param <T> the type of the input elements
     * @param generator a function which produces a new array of the desired
     *        type and the provided length
     * @return a {@code Collector} which collects all the input elements into an
     *         array, in encounter order
     * @throws NullPointerException if the generator is null.
     */
    public static <T extends @Nullable Object> Collector<T, ?, T[]> toArray(IntFunction<T[]> generator) {
        Objects.requireNonNull(generator);
        return Collectors.collectingAndThen(Collectors.toList(), list -> list.toArray(generator.apply(list.size())));
    }

    /**
     * Returns a {@code Collector} which produces a boolean array containing the
     * results of applying the given predicate to the input elements, in
     * encounter order.
     * 
     * @param <T> the type of the input elements
     * @param predicate a non-interfering, stateless predicate to apply to each
     *        input element. The result values of this predicate are collected
     *        to the resulting boolean array.
     * @return a {@code Collector} which collects the results of the predicate
     *         function to the boolean array, in encounter order.
     * @throws NullPointerException if predicate is null.
     * @since 0.3.8
     */
    public static <T extends @Nullable Object> Collector<T, ?, boolean[]> toBooleanArray(Predicate<? super T> predicate) {
        Objects.requireNonNull(predicate);
        return PartialCollector.booleanArray().asRef((box, t) -> {
            if (predicate.test(t))
                box.a.set(box.b);
            box.b = StrictMath.addExact(box.b, 1);
        });
    }

    /**
     * Returns a {@code Collector} that accumulates the input enum values into a
     * new {@code EnumSet}.
     *
     * <p>
     * This method returns a
     * <a href="package-summary.html#ShortCircuitReduction">short-circuiting
     * collector</a>: it may not process all the elements if the resulting set
     * contains all possible enum values.
     * 
     * @param <T> the type of the input elements
     * @param enumClass the class of input enum values
     * @return a {@code Collector} which collects all the input elements into a
     *         {@code EnumSet}
     * @throws NullPointerException if enumClass is null.
     */
    public static <T extends Enum<T>> Collector<T, ?, EnumSet<T>> toEnumSet(Class<T> enumClass) {
        int size = EnumSet.allOf(enumClass).size();
        return new CancellableCollectorImpl<>(() -> EnumSet.noneOf(enumClass), EnumSet::add, (s1, s2) -> {
            s1.addAll(s2);
            return s1;
        }, Function.identity(), set -> set.size() == size, UNORDERED_ID_CHARACTERISTICS);
    }

    /**
     * Returns a {@code Collector} that accumulates elements into a {@code Map}
     * whose keys and values are taken from {@code Map.Entry}.
     *
     * <p>
     * There are no guarantees on the type or serializability of the {@code Map} returned;
     * if more control over the returned {@code Map} is required, use {@link #entriesToCustomMap(Supplier)}
     *
     * <p>
     * Returned {@code Map} is guaranteed to be modifiable. See {@link one.util.streamex.EntryStream#toMap()}.
     *
     * @param <K> the type of the map keys
     * @param <V> the type of the map values
     * @return {@code Collector} which collects elements into a {@code Map}
     * whose keys and values are taken from {@code Map.Entry}
     * @throws IllegalStateException if this stream contains duplicate keys
     *                               (according to {@link Object#equals(Object)}).
     * @see #entriesToMap(BinaryOperator)
     * @see Collectors#toMap(Function, Function)
     * @since 0.7.3
     */
    public static <K extends @Nullable Object, V extends @Nullable Object> Collector<Entry<? extends K, ? extends V>, ?, Map<K, V>> entriesToMap() {
        return entriesToCustomMap(HashMap::new);
    }

    /**
     * Returns a {@code Collector} that accumulates elements into a {@code Map}
     * whose keys and values are taken from {@code Map.Entry} and combining them
     * using the provided {@code combiner} function to the input elements.
     *
     * <p>
     * There are no guarantees on the type or serializability of the {@code Map} returned;
     * if more control over the returned {@code Map} is required, use {@link #entriesToCustomMap(BinaryOperator, Supplier)}
     *
     * <p>
     * Returned {@code Map} is guaranteed to be modifiable. See {@link one.util.streamex.EntryStream#toMap()}.
     *
     * <p>If the mapped keys contain duplicates (according to {@link Object#equals(Object)}),
     * the value mapping function is applied to each equal element, and the
     * results are merged using the provided {@code combiner} function.
     *
     * @param <K>      the type of the map keys
     * @param <V>      the type of the map values
     * @param combiner a merge function, used to resolve collisions between
     *                 values associated with the same key, as supplied
     *                 to {@link Map#merge(Object, Object, BiFunction)}
     * @return {@code Collector} which collects elements into a {@code Map}
     * whose keys and values are taken from {@code Map.Entry} and combining them
     * using the {@code combiner} function
     * @throws NullPointerException if the combiner is null.
     * @see #entriesToMap()
     * @see Collectors#toMap(Function, Function, BinaryOperator)
     * @since 0.7.3
     */
    public static <K extends @Nullable Object, V extends @Nullable Object> Collector<Entry<? extends K, ? extends V>, ?, Map<K, V>> entriesToMap(
            BinaryOperator<V> combiner) {
        return entriesToCustomMap(combiner, HashMap::new);
    }

    /**
     * Returns a {@code Collector} that accumulates elements into
     * a result {@code Map} defined by {@code mapSupplier} function
     * whose keys and values are taken from {@code Map.Entry}.
     *
     * @param <K> the type of the map keys
     * @param <V> the type of the map values
     * @param <M> the type of the resulting {@code Map}
     * @param mapSupplier a function which returns a new, empty {@code Map} into
     *                    which the results will be inserted
     * @return {@code Collector} which collects elements into a {@code Map}
     * defined by {@code mapSupplier} function
     * whose keys and values are taken from {@code Map.Entry}
     * @throws IllegalStateException if this stream contains duplicate keys
     *                               (according to {@link Object#equals(Object)}).
     * @throws NullPointerException  if mapSupplier is null.
     * @throws NullPointerException  if entry value is null.
     * @see #entriesToCustomMap(BinaryOperator, Supplier)
     * @see Collector#of(Supplier, BiConsumer, BinaryOperator, Collector.Characteristics...)
     * @since 0.7.3
     */
    public static <
            K extends @Nullable Object,
            V,
            M extends Map<K, V>> Collector<Entry<? extends K, ? extends V>, ?, M> entriesToCustomMap(
            Supplier<M> mapSupplier) {
        return Collector.of(mapSupplier,
                (m, entry) -> addToMap(m, entry.getKey(), Objects.requireNonNull(entry.getValue())),
                (m1, m2) -> {
                    m2.forEach((k, v) -> addToMap(m1, k, v));
                    return m1;
                }
        );
    }

    /**
     * Returns a {@code Collector} that accumulates elements into
     * a result {@code Map} defined by {@code mapSupplier} function
     * whose keys and values are taken from {@code Map.Entry} and combining them
     * using the provided {@code combiner} function to the input elements.
     *
     * <p>If the mapped keys contain duplicates (according to {@link Object#equals(Object)}),
     * the value mapping function is applied to each equal element, and the
     * results are merged using the provided {@code combiner} function.
     *
     * @param <K>         the type of the map keys
     * @param <V>         the type of the map values
     * @param <M>         the type of the resulting {@code Map}
     * @param combiner    a merge function, used to resolve collisions between
     *                    values associated with the same key, as supplied
     *                    to {@link Map#merge(Object, Object, BiFunction)}
     * @param mapSupplier a function which returns a new, empty {@code Map} into
     *                    which the results will be inserted
     * @return {@code Collector} which collects elements into a {@code Map}
     * whose keys and values are taken from {@code Map.Entry} and combining them
     * using the {@code combiner} function
     * @throws NullPointerException if {@code combiner} is null.
     * @throws NullPointerException if {@code mapSupplier} is null.
     * @see #entriesToCustomMap(Supplier)
     * @see Collectors#toMap(Function, Function, BinaryOperator, Supplier)
     * @since 0.7.3
     */
    public static <
            K extends @Nullable Object,
            V extends @Nullable Object,
            M extends Map<K, V>> Collector<Entry<? extends K, ? extends V>, ?, M> entriesToCustomMap(
            BinaryOperator<V> combiner, Supplier<M> mapSupplier) {
        Objects.requireNonNull(combiner);
        Objects.requireNonNull(mapSupplier);
        return Collectors.toMap(Entry::getKey, Entry::getValue, combiner, mapSupplier);
    }

    /**
     * Returns a {@code Collector} which counts a number of distinct values the
     * mapper function returns for the stream elements.
     * 
     * <p>
     * The operation performed by the returned collector is equivalent to
     * {@code stream.map(mapper).distinct().count()}. This collector is mostly
     * useful as a downstream collector.
     * 
     * @param <T> the type of the input elements
     * @param mapper a function which classifies input elements.
     * @return a collector which counts a number of distinct classes, the mapper
     *         function returns for the stream elements.
     * @throws NullPointerException if mapper is null.
     */
    public static <T extends @Nullable Object> Collector<T, ?, Integer> distinctCount(Function<? super T, ?> mapper) {
        Objects.requireNonNull(mapper);
        return Collectors.collectingAndThen(Collectors.mapping(mapper, Collectors.toSet()), Set::size);
    }

    /**
     * Returns a {@code Collector} which collects into the {@link List} the
     * input elements for which given mapper function returns distinct results.
     *
     * <p>
     * For an ordered source the order of collected elements is preserved. If the
     * same result is returned by the mapper function for several elements, only the
     * first element is included in the resulting list.
     * 
     * <p>
     * There are no guarantees on the type, mutability, serializability, or
     * thread-safety of the {@code List} returned.
     * 
     * <p>
     * The operation performed by the returned collector is equivalent to
     * {@code stream.distinct(mapper).toList()}, but may work faster.
     * 
     * @param <T> the type of the input elements
     * @param mapper a function which classifies input elements.
     * @return a collector which collects distinct elements to the {@code List}.
     * @throws NullPointerException if mapper is null.
     * @since 0.3.8
     */
    public static <T extends @Nullable Object> Collector<T, ?, List<T>> distinctBy(Function<? super T, ?> mapper) {
        Objects.requireNonNull(mapper);
        return Collector.<T, Map<Object, T>, List<T>>of(LinkedHashMap::new, (map, t) -> map.putIfAbsent(mapper.apply(
            t), t), (m1, m2) -> {
                for (Entry<Object, T> e : m2.entrySet()) {
                    m1.putIfAbsent(e.getKey(), e.getValue());
                }
                return m1;
            }, map -> new ArrayList<>(map.values()));
    }

    /**
     * Returns a {@code Collector} accepting elements of type {@code T} that
     * counts the number of input elements and returns result as {@code Integer}
     * . If no elements are present, the result is 0.
     *
     * @param <T> the type of the input elements
     * @return a {@code Collector} that counts the input elements
     * @since 0.3.3
     * @see Collectors#counting()
     */
    public static <T extends @Nullable Object> Collector<T, ?, Integer> countingInt() {
        return PartialCollector.intSum().asRef((acc, t) -> acc[0]++);
    }

    /**
     * Returns a {@code Collector} which aggregates the results of two supplied
     * collectors using the supplied finisher function.
     * 
     * <p>
     * This method returns a
     * <a href="package-summary.html#ShortCircuitReduction">short-circuiting
     * collector</a> if both downstream collectors are short-circuiting. The
     * collection might stop when both downstream collectors report that the
     * collection is complete.
     * 
     * <p>
     * This collector is similar to the {@code teeing} collector available since
     * JDK 12. The only difference is that this collector correctly combines 
     * short-circuiting collectors.
     * 
     * @param <T> the type of the input elements
     * @param <A1> the intermediate accumulation type of the first collector
     * @param <A2> the intermediate accumulation type of the second collector
     * @param <R1> the result type of the first collector
     * @param <R2> the result type of the second collector
     * @param <R> the final result type
     * @param c1 the first collector
     * @param c2 the second collector
     * @param finisher the function which merges two results into the single
     *        one.
     * @return a {@code Collector} which aggregates the results of two supplied
     *         collectors.
     * @throws NullPointerException if c1 is null, or c2 is null, or finisher is null.
     */
    public static <
            T extends @Nullable Object,
            A1 extends @Nullable Object,
            A2 extends @Nullable Object,
            R1 extends @Nullable Object,
            R2 extends @Nullable Object,
            R extends @Nullable Object> Collector<T, ?, R> pairing(
            Collector<? super T, A1, R1> c1,
            Collector<? super T, A2, R2> c2,
            BiFunction<? super R1, ? super R2, ? extends R> finisher) {
        Objects.requireNonNull(finisher);
        EnumSet<Characteristics> c = EnumSet.noneOf(Characteristics.class);
        c.addAll(c1.characteristics());
        c.retainAll(c2.characteristics());
        c.remove(Characteristics.IDENTITY_FINISH);

        Supplier<A1> c1Supplier = c1.supplier();
        Supplier<A2> c2Supplier = c2.supplier();
        BiConsumer<A1, ? super T> c1Accumulator = c1.accumulator();
        BiConsumer<A2, ? super T> c2Accumulator = c2.accumulator();
        BinaryOperator<A1> c1Combiner = c1.combiner();
        BinaryOperator<A2> c2combiner = c2.combiner();

        Supplier<PairBox<A1, A2>> supplier = () -> new PairBox<>(c1Supplier.get(), c2Supplier.get());
        BiConsumer<PairBox<A1, A2>, T> accumulator = (acc, v) -> {
            c1Accumulator.accept(acc.a, v);
            c2Accumulator.accept(acc.b, v);
        };
        BinaryOperator<PairBox<A1, A2>> combiner = (acc1, acc2) -> {
            acc1.a = c1Combiner.apply(acc1.a, acc2.a);
            acc1.b = c2combiner.apply(acc1.b, acc2.b);
            return acc1;
        };
        Function<PairBox<A1, A2>, R> resFinisher = acc -> {
            R1 r1 = c1.finisher().apply(acc.a);
            R2 r2 = c2.finisher().apply(acc.b);
            return finisher.apply(r1, r2);
        };
        Predicate<A1> c1Finished = finished(c1);
        Predicate<A2> c2Finished = finished(c2);
        if (c1Finished != null && c2Finished != null) {
            Predicate<PairBox<A1, A2>> finished = acc -> c1Finished.test(acc.a) && c2Finished.test(acc.b);
            return new CancellableCollectorImpl<>(supplier, accumulator, combiner, resFinisher, finished, c);
        }
        return Collector.of(supplier, accumulator, combiner, resFinisher, c.toArray(new Characteristics[0]));
    }

    /**
     * Returns a {@code Collector} which finds the minimal and maximal element
     * according to the supplied comparator, then applies finisher function to
     * them producing the final result.
     * 
     * <p>
     * This collector produces the stable result for an ordered stream: if several
     * minimal or maximal elements appear, the collector always selects the
     * first encountered.
     * 
     * <p>
     * If there are no input elements, the finisher method is not called and
     * empty {@code Optional} is returned. Otherwise, the finisher result is
     * wrapped into {@code Optional}.
     *
     * @param <T> the type of the input elements
     * @param <R> the type of the result wrapped into {@code Optional}
     * @param comparator comparator which is used to find the minimal and the maximal
     *        element
     * @param finisher a {@link BiFunction} which takes the minimal and the maximal
     *        element and produces the final result.
     * @return a {@code Collector} which finds minimal and maximal elements.
     * @throws NullPointerException if the comparator is null, the finisher is null,
     * or the finisher returns null.
     */
    public static <T extends @Nullable Object, R> Collector<T, ?, Optional<R>> minMax(Comparator<? super T> comparator,
            BiFunction<? super T, ? super T, ? extends R> finisher) {
        Objects.requireNonNull(finisher);
        return pairing(Collectors.minBy(comparator), Collectors.maxBy(comparator),
            (min, max) -> min.isPresent() ? Optional.of(finisher.apply(min.get(), max.get())) : Optional.empty());
    }

    /**
     * Returns a {@code Collector} which finds all the elements which are equal
     * to each other and bigger than any other element according to the
     * specified {@link Comparator}. The found elements are reduced using the
     * specified downstream {@code Collector}.
     *
     * @param <T> the type of the input elements
     * @param <A> the intermediate accumulation type of the downstream collector
     * @param <D> the result type of the downstream reduction
     * @param comparator a {@code Comparator} to compare the elements
     * @param downstream a {@code Collector} implementing the downstream
     *        reduction
     * @return a {@code Collector} which finds all the maximal elements.
     * @throws NullPointerException if comparator is null, or downstream is null.
     * @see #maxAll(Comparator)
     * @see #maxAll(Collector)
     * @see #maxAll()
     */
    public static <
            T extends @Nullable Object,
            A extends @Nullable Object,
            D extends @Nullable Object> Collector<T, ?, D> maxAll(
            Comparator<? super T> comparator,
            Collector<? super T, A, D> downstream) {
        Objects.requireNonNull(comparator);
        Supplier<A> downstreamSupplier = downstream.supplier();
        BiConsumer<A, ? super T> downstreamAccumulator = downstream.accumulator();
        BinaryOperator<A> downstreamCombiner = downstream.combiner();
        Supplier<PairBox<A, T>> supplier = () -> new PairBox<>(downstreamSupplier.get(), none());
        BiConsumer<PairBox<A, T>, T> accumulator = (acc, t) -> {
            if (acc.b == NONE) {
                downstreamAccumulator.accept(acc.a, t);
                acc.b = t;
            } else {
                int cmp = comparator.compare(t, acc.b);
                if (cmp > 0) {
                    acc.a = downstreamSupplier.get();
                    acc.b = t;
                }
                if (cmp >= 0)
                    downstreamAccumulator.accept(acc.a, t);
            }
        };
        BinaryOperator<PairBox<A, T>> combiner = (acc1, acc2) -> {
            if (acc2.b == NONE) {
                return acc1;
            }
            if (acc1.b == NONE) {
                return acc2;
            }
            int cmp = comparator.compare(acc1.b, acc2.b);
            if (cmp > 0) {
                return acc1;
            }
            if (cmp < 0) {
                return acc2;
            }
            acc1.a = downstreamCombiner.apply(acc1.a, acc2.a);
            return acc1;
        };
        Function<PairBox<A, T>, D> finisher = acc -> downstream.finisher().apply(acc.a);
        return Collector.of(supplier, accumulator, combiner, finisher);
    }

    /**
     * Returns a {@code Collector} which finds all the elements which are equal
     * to each other and bigger than any other element according to the
     * specified {@link Comparator}. The found elements are collected to
     * {@link List}.
     *
     * @param <T> the type of the input elements
     * @param comparator a {@code Comparator} to compare the elements
     * @return a {@code Collector} which finds all the maximal elements and
     *         collects them to the {@code List}.
     * @throws NullPointerException if the comparator is null.
     * @see #maxAll(Comparator, Collector)
     * @see #maxAll()
     */
    public static <T extends @Nullable Object> Collector<T, ?, List<T>> maxAll(Comparator<? super T> comparator) {
        return maxAll(comparator, Collectors.toList());
    }

    /**
     * Returns a {@code Collector} which finds all the elements which are equal
     * to each other and bigger than any other element according to the natural
     * order. The found elements are reduced using the specified downstream
     * {@code Collector}.
     *
     * @param <T> the type of the input elements
     * @param <A> the intermediate accumulation type of the downstream collector
     * @param <D> the result type of the downstream reduction
     * @param downstream a {@code Collector} implementing the downstream
     *        reduction
     * @return a {@code Collector} which finds all the maximal elements.
     * @throws NullPointerException if downstream is null.
     * @see #maxAll(Comparator, Collector)
     * @see #maxAll(Comparator)
     * @see #maxAll()
     */
    public static <
            T extends Comparable<? super T>,
            A extends @Nullable Object,
            D extends @Nullable Object> Collector<T, ?, D> maxAll(Collector<T, A, D> downstream) {
        return maxAll(Comparator.<T>naturalOrder(), downstream);
    }

    /**
     * Returns a {@code Collector} which finds all the elements which are equal
     * to each other and bigger than any other element according to the natural
     * order. The found elements are collected to {@link List}.
     *
     * @param <T> the type of the input elements
     * @return a {@code Collector} which finds all the maximal elements and
     *         collects them to the {@code List}.
     * @see #maxAll(Comparator)
     * @see #maxAll(Collector)
     */
    public static <T extends Comparable<? super T>> Collector<T, ?, List<T>> maxAll() {
        return maxAll(Comparator.naturalOrder(), Collectors.toList());
    }

    /**
     * Returns a {@code Collector} which finds all the elements which are equal
     * to each other and smaller than any other element according to the
     * specified {@link Comparator}. The found elements are reduced using the
     * specified downstream {@code Collector}.
     *
     * @param <T> the type of the input elements
     * @param <A> the intermediate accumulation type of the downstream collector
     * @param <D> the result type of the downstream reduction
     * @param comparator a {@code Comparator} to compare the elements
     * @param downstream a {@code Collector} implementing the downstream
     *        reduction
     * @return a {@code Collector} which finds all the minimal elements.
     * @throws NullPointerException if comparator is null, or downstream is null.
     * @see #minAll(Comparator)
     * @see #minAll(Collector)
     * @see #minAll()
     */
    public static <
            T extends @Nullable Object,
            A extends @Nullable Object,
            D extends @Nullable Object> Collector<T, ?, D> minAll(
            Comparator<? super T> comparator, Collector<T, A, D> downstream) {
        return maxAll(comparator.reversed(), downstream);
    }

    /**
     * Returns a {@code Collector} which finds all the elements which are equal
     * to each other and smaller than any other element according to the
     * specified {@link Comparator}. The found elements are collected to
     * {@link List}.
     *
     * @param <T> the type of the input elements
     * @param comparator a {@code Comparator} to compare the elements
     * @return a {@code Collector} which finds all the minimal elements and
     *         collects them to the {@code List}.
     * @throws NullPointerException if the comparator is null.
     * @see #minAll(Comparator, Collector)
     * @see #minAll()
     */
    public static <T extends @Nullable Object> Collector<T, ?, List<T>> minAll(Comparator<? super T> comparator) {
        return maxAll(comparator.reversed(), Collectors.toList());
    }

    /**
     * Returns a {@code Collector} which finds all the elements which are equal
     * to each other and smaller than any other element according to the natural
     * order. The found elements are reduced using the specified downstream
     * {@code Collector}.
     *
     * @param <T> the type of the input elements
     * @param <A> the intermediate accumulation type of the downstream collector
     * @param <D> the result type of the downstream reduction
     * @param downstream a {@code Collector} implementing the downstream
     *        reduction
     * @return a {@code Collector} which finds all the minimal elements.
     * @throws NullPointerException if downstream is null.
     * @see #minAll(Comparator, Collector)
     * @see #minAll(Comparator)
     * @see #minAll()
     */
    public static <T extends Comparable<? super T>,
            A extends @Nullable Object,
            D extends @Nullable Object> Collector<T, ?, D> minAll(Collector<T, A, D> downstream) {
        return maxAll(Comparator.<T>reverseOrder(), downstream);
    }

    /**
     * Returns a {@code Collector} which finds all the elements which are equal
     * to each other and smaller than any other element according to the natural
     * order. The found elements are collected to {@link List}.
     *
     * @param <T> the type of the input elements
     * @return a {@code Collector} which finds all the minimal elements and
     *         collects them to the {@code List}.
     * @see #minAll(Comparator)
     * @see #minAll(Collector)
     */
    public static <T extends Comparable<? super T>> Collector<T, ?, List<T>> minAll() {
        return maxAll(Comparator.reverseOrder(), Collectors.toList());
    }

    /**
     * Returns a {@code Collector} which collects the stream element if stream
     * contains exactly one element.
     * 
     * <p>
     * This method returns a
     * <a href="package-summary.html#ShortCircuitReduction">short-circuiting
     * collector</a>.
     * 
     * @param <T> the type of the input elements
     * @return a collector which returns an {@link Optional} describing the only
     *         element of the stream. For empty stream or stream containing more
     *         than one element an empty {@code Optional} is returned.
     * @throws NullPointerException if the only stream element is null.
     * @since 0.4.0
     */
    public static <T> Collector<T, ?, Optional<T>> onlyOne() {
        return new CancellableCollectorImpl<T, Box<Optional<T>>, Optional<T>>(Box::new, (box,
                t) -> box.a = box.a == null ? Optional.of(t) : Optional.empty(), (box1, box2) -> box1.a == null ? box2
                        : box2.a == null ? box1 : new Box<>(Optional.empty()), box -> box.a == null ? Optional.empty()
                                : box.a, box -> box.a != null && !box.a.isPresent(), UNORDERED_CHARACTERISTICS);
    }

    /**
     * Returns a {@code Collector} which collects the stream element satisfying the predicate
     * if there is only one such element.
     *
     * <p>
     * This method returns a
     * <a href="package-summary.html#ShortCircuitReduction">short-circuiting
     * collector</a>.
     *
     * @param predicate a predicate to be applied to the stream elements
     * @param <T> the type of the input elements
     * @return a collector which returns an {@link Optional} describing the only
     *         element of the stream satisfying the predicate. If stream contains no elements satisfying the predicate,
     *         or more than one such element, an empty {@code Optional} is returned.
     * @throws NullPointerException if the predicate is null or the only stream element is null.
     * @since 0.6.7
     */
    public static <T> Collector<T, ?, Optional<T>> onlyOne(Predicate<? super T> predicate) {
        return filtering(predicate, onlyOne());
    }

    /**
     * Returns a {@code Collector} which collects only the first stream element
     * if any.
     * 
     * <p>
     * This method returns a
     * <a href="package-summary.html#ShortCircuitReduction">short-circuiting
     * collector</a>.
     * 
     * <p>
     * The operation performed by the returned collector is equivalent to
     * {@code stream.findFirst()}. This collector is mostly useful as a
     * downstream collector.
     * 
     * @param <T> the type of the input elements
     * @return a collector which returns an {@link Optional} which describes the
     *         first element of the stream. For empty stream an empty
     *         {@code Optional} is returned.
     * @throws NullPointerException if the first stream element is null.
     */
    public static <T> Collector<T, ?, Optional<T>> first() {
        return new CancellableCollectorImpl<>(() -> new Box<T>(none()), (box, t) -> {
            if (box.a == NONE)
                box.a = t;
        }, (box1, box2) -> box1.a == NONE ? box2 : box1, box -> box.a == NONE ? Optional.empty() : Optional.of(box.a),
                box -> box.a != NONE, NO_CHARACTERISTICS);
    }

    /**
