/*

 * Copyright (c) 2012, 2015, Oracle and/or its affiliates. All rights reserved.

 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.

 *

 * This code is free software; you can redistribute it and/or modify it

 * under the terms of the GNU General Public License version 2 only, as

 * published by the Free Software Foundation.  Oracle designates this

 * particular file as subject to the "Classpath" exception as provided

 * by Oracle in the LICENSE file that accompanied this code.

 *

 * This code is distributed in the hope that it will be useful, but WITHOUT

 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or

 * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License

 * version 2 for more details (a copy is included in the LICENSE file that

 * accompanied this code).

 *

 * You should have received a copy of the GNU General Public License version

 * 2 along with this work; if not, write to the Free Software Foundation,

 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.

 *

 * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA

 * or visit www.oracle.com if you need additional information or have any

 * questions.

 */

package java.util.stream;

import java.util.AbstractMap;

import java.util.AbstractSet;

import java.util.ArrayList;

import java.util.Arrays;

import java.util.Collection;

import java.util.Collections;

import java.util.Comparator;

import java.util.DoubleSummaryStatistics;

import java.util.EnumSet;

import java.util.HashMap;

import java.util.HashSet;

import java.util.IntSummaryStatistics;

import java.util.Iterator;

import java.util.List;

import java.util.LongSummaryStatistics;

import java.util.Map;

import java.util.Objects;

import java.util.Optional;

import java.util.Set;

import java.util.StringJoiner;

import java.util.concurrent.ConcurrentHashMap;

import java.util.concurrent.ConcurrentMap;

import java.util.function.BiConsumer;

import java.util.function.BiFunction;

import java.util.function.BinaryOperator;

import java.util.function.Consumer;

import java.util.function.Function;

import java.util.function.Predicate;

import java.util.function.Supplier;

import java.util.function.ToDoubleFunction;

import java.util.function.ToIntFunction;

import java.util.function.ToLongFunction;

/**

 * Implementations of {@link Collector} that implement various useful reduction

 * operations, such as accumulating elements into collections, summarizing

 * elements according to various criteria, etc.

 *

 * <p>The following are examples of using the predefined collectors to perform

 * common mutable reduction tasks:

 *

 * <pre>{@code

 *     // Accumulate names into a List

 *     List<String> list = people.stream().map(Person::getName).collect(Collectors.toList());

 *

 *     // Accumulate names into a TreeSet

 *     Set<String> set = people.stream().map(Person::getName).collect(Collectors.toCollection(TreeSet::new));

 *

 *     // Convert elements to strings and concatenate them, separated by commas

 *     String joined = things.stream()

 *                           .map(Object::toString)

 *                           .collect(Collectors.joining(", "));

 *

 *     // Compute sum of salaries of employee

 *     int total = employees.stream()

 *                          .collect(Collectors.summingInt(Employee::getSalary)));

 *

 *     // Group employees by department

 *     Map<Department, List<Employee>> byDept

 *         = employees.stream()

 *                    .collect(Collectors.groupingBy(Employee::getDepartment));

 *

 *     // Compute sum of salaries by department

 *     Map<Department, Integer> totalByDept

 *         = employees.stream()

 *                    .collect(Collectors.groupingBy(Employee::getDepartment,

 *                                                   Collectors.summingInt(Employee::getSalary)));

 *

 *     // Partition students into passing and failing

 *     Map<Boolean, List<Student>> passingFailing =

 *         students.stream()

 *                 .collect(Collectors.partitioningBy(s -> s.getGrade() >= PASS_THRESHOLD));

 *

 * }</pre>

 *

 * @since 1.8

 */

public final class Collectors {

    static final Set<Collector.Characteristics> CH_CONCURRENT_ID

            = Collections.unmodifiableSet(EnumSet.of(Collector.Characteristics.CONCURRENT,

                                                     Collector.Characteristics.UNORDERED,

                                                     Collector.Characteristics.IDENTITY_FINISH));

    static final Set<Collector.Characteristics> CH_CONCURRENT_NOID

            = Collections.unmodifiableSet(EnumSet.of(Collector.Characteristics.CONCURRENT,

                                                     Collector.Characteristics.UNORDERED));

    static final Set<Collector.Characteristics> CH_ID

            = Collections.unmodifiableSet(EnumSet.of(Collector.Characteristics.IDENTITY_FINISH));

    static final Set<Collector.Characteristics> CH_UNORDERED_ID

            = Collections.unmodifiableSet(EnumSet.of(Collector.Characteristics.UNORDERED,

                                                     Collector.Characteristics.IDENTITY_FINISH));

    static final Set<Collector.Characteristics> CH_NOID = Collections.emptySet();

    private Collectors() { }

    /**

     * Construct an {@code IllegalStateException} with appropriate message.

     *

     * @param k the duplicate key

     * @param u 1st value to be accumulated/merged

     * @param v 2nd value to be accumulated/merged

     */

    private static IllegalStateException duplicateKeyException(

            Object k, Object u, Object v) {

        return new IllegalStateException(String.format(

            "Duplicate key %s (attempted merging values %s and %s)",

            k, u, v));

    }

    /**

     * {@code BinaryOperator<Map>} that merges the contents of its right

     * argument into its left argument, throwing {@code IllegalStateException}

     * if duplicate keys are encountered.

     *

     * @param <K> type of the map keys

     * @param <V> type of the map values

     * @param <M> type of the map

     * @return a merge function for two maps

     */

    private static <K, V, M extends Map<K,V>>

    BinaryOperator<M> uniqKeysMapMerger() {

        return (m1, m2) -> {

            for (Map.Entry<K,V> e : m2.entrySet()) {

                K k = e.getKey();

                V v = Objects.requireNonNull(e.getValue());

                V u = m1.putIfAbsent(k, v);

                if (u != null) throw duplicateKeyException(k, u, v);

            }

            return m1;

        };

    }

    /**

     * {@code BiConsumer<Map, T>} that accumulates (key, value) pairs

     * extracted from elements into the map, throwing {@code IllegalStateException}

     * if duplicate keys are encountered.

     *

     * @param keyMapper a function that maps an element into a key

     * @param valueMapper a function that maps an element into a value

     * @param <T> type of elements

     * @param <K> type of map keys

     * @param <V> type of map values

     * @return an accumulating consumer

     */

    private static <T, K, V>

    BiConsumer<Map<K, V>, T> uniqKeysMapAccumulator(Function<? super T, ? extends K> keyMapper,

                                                    Function<? super T, ? extends V> valueMapper) {

        return (map, element) -> {

            K k = keyMapper.apply(element);

            V v = Objects.requireNonNull(valueMapper.apply(element));

            V u = map.putIfAbsent(k, v);

            if (u != null) throw duplicateKeyException(k, u, v);

        };

    }

    @SuppressWarnings("unchecked")

    private static <I, R> Function<I, R> castingIdentity() {

        return i -> (R) i;

    }

    /**

     * Simple implementation class for {@code Collector}.

     *

     * @param <T> the type of elements to be collected

     * @param <R> the type of the result

     */

    static class CollectorImpl<T, A, R> implements Collector<T, A, R> {

        private final Supplier<A> supplier;

        private final BiConsumer<A, T> accumulator;

        private final BinaryOperator<A> combiner;

        private final Function<A, R> finisher;

        private final Set<Characteristics> characteristics;

        CollectorImpl(Supplier<A> supplier,

                      BiConsumer<A, T> accumulator,

                      BinaryOperator<A> combiner,

                      Function<A,R> finisher,

                      Set<Characteristics> characteristics) {

            this.supplier = supplier;

            this.accumulator = accumulator;

            this.combiner = combiner;

            this.finisher = finisher;

            this.characteristics = characteristics;

        }

        CollectorImpl(Supplier<A> supplier,

                      BiConsumer<A, T> accumulator,

                      BinaryOperator<A> combiner,

                      Set<Characteristics> characteristics) {

            this(supplier, accumulator, combiner, castingIdentity(), characteristics);

        }

        @Override

        public BiConsumer<A, T> accumulator() {

            return accumulator;

        }

        @Override

        public Supplier<A> supplier() {

            return supplier;

        }

        @Override

        public BinaryOperator<A> combiner() {

            return combiner;

        }

        @Override

        public Function<A, R> finisher() {

            return finisher;

        }

        @Override

        public Set<Characteristics> characteristics() {

            return characteristics;

        }

    }

    /**

     * Returns a {@code Collector} that accumulates the input elements into a

     * new {@code Collection}, in encounter order.  The {@code Collection} is

     * created by the provided factory.

     *

     * @param <T> the type of the input elements

     * @param <C> the type of the resulting {@code Collection}

     * @param collectionFactory a {@code Supplier} which returns a new, empty

     * {@code Collection} of the appropriate type

     * @return a {@code Collector} which collects all the input elements into a

     * {@code Collection}, in encounter order

     */

    public static <T, C extends Collection<T>>

    Collector<T, ?, C> toCollection(Supplier<C> collectionFactory) {

        return new CollectorImpl<>(collectionFactory, Collection<T>::add,

                                   (r1, r2) -> { r1.addAll(r2); return r1; },

                                   CH_ID);

    }

    /**

     * Returns a {@code Collector} that accumulates the input elements into a

     * new {@code List}. There are no guarantees on the type, mutability,

     * serializability, or thread-safety of the {@code List} returned; if more

     * control over the returned {@code List} is required, use {@link #toCollection(Supplier)}.

     *

     * @param <T> the type of the input elements

     * @return a {@code Collector} which collects all the input elements into a

     * {@code List}, in encounter order

     */

    public static <T>

    Collector<T, ?, List<T>> toList() {

        return new CollectorImpl<>((Supplier<List<T>>) ArrayList::new, List::add,

                                   (left, right) -> { left.addAll(right); return left; },

                                   CH_ID);

    }

    /**

     * Returns a {@code Collector} that accumulates the input elements into a

     * new {@code Set}. There are no guarantees on the type, mutability,

     * serializability, or thread-safety of the {@code Set} returned; if more

     * control over the returned {@code Set} is required, use

     * {@link #toCollection(Supplier)}.

     *

     * <p>This is an {@link Collector.Characteristics#UNORDERED unordered}

     * Collector.

     *

     * @param <T> the type of the input elements

     * @return a {@code Collector} which collects all the input elements into a

     * {@code Set}

     */

    public static <T>

    Collector<T, ?, Set<T>> toSet() {

        return new CollectorImpl<>((Supplier<Set<T>>) HashSet::new, Set::add,

                                   (left, right) -> { left.addAll(right); return left; },

                                   CH_UNORDERED_ID);

    }

    /**

     * Returns a {@code Collector} that concatenates the input elements into a

     * {@code String}, in encounter order.

     *

     * @return a {@code Collector} that concatenates the input elements into a

     * {@code String}, in encounter order

     */

    public static Collector<CharSequence, ?, String> joining() {

        return new CollectorImpl<CharSequence, StringBuilder, String>(

                StringBuilder::new, StringBuilder::append,

                (r1, r2) -> { r1.append(r2); return r1; },

                StringBuilder::toString, CH_NOID);

    }

    /**

     * Returns a {@code Collector} that concatenates the input elements,

     * separated by the specified delimiter, in encounter order.

     *

     * @param delimiter the delimiter to be used between each element

     * @return A {@code Collector} which concatenates CharSequence elements,

     * separated by the specified delimiter, in encounter order

     */

    public static Collector<CharSequence, ?, String> joining(CharSequence delimiter) {

        return joining(delimiter, "", "");

    }

    /**

     * Returns a {@code Collector} that concatenates the input elements,

     * separated by the specified delimiter, with the specified prefix and

     * suffix, in encounter order.

     *

     * @param delimiter the delimiter to be used between each element

     * @param  prefix the sequence of characters to be used at the beginning

     *                of the joined result

     * @param  suffix the sequence of characters to be used at the end

     *                of the joined result

     * @return A {@code Collector} which concatenates CharSequence elements,

     * separated by the specified delimiter, in encounter order

     */

    public static Collector<CharSequence, ?, String> joining(CharSequence delimiter,

                                                             CharSequence prefix,

                                                             CharSequence suffix) {

        return new CollectorImpl<>(

                () -> new StringJoiner(delimiter, prefix, suffix),

                StringJoiner::add, StringJoiner::merge,

                StringJoiner::toString, CH_NOID);

    }

    /**

     * {@code BinaryOperator<Map>} that merges the contents of its right

     * argument into its left argument, using the provided merge function to

     * handle duplicate keys.

     *

     * @param <K> type of the map keys

     * @param <V> type of the map values

     * @param <M> type of the map

     * @param mergeFunction A merge function suitable for

     * {@link Map#merge(Object, Object, BiFunction) Map.merge()}

     * @return a merge function for two maps

     */

    private static <K, V, M extends Map<K,V>>

    BinaryOperator<M> mapMerger(BinaryOperator<V> mergeFunction) {

        return (m1, m2) -> {

            for (Map.Entry<K,V> e : m2.entrySet())

                m1.merge(e.getKey(), e.getValue(), mergeFunction);

            return m1;

        };

    }

    /**

     * Adapts a {@code Collector} accepting elements of type {@code U} to one

     * accepting elements of type {@code T} by applying a mapping function to

     * each input element before accumulation.

     *

     * @apiNote

     * The {@code mapping()} collectors are most useful when used in a

     * multi-level reduction, such as downstream of a {@code groupingBy} or

     * {@code partitioningBy}.  For example, given a stream of

     * {@code Person}, to accumulate the set of last names in each city:

     * <pre>{@code

     *     Map<City, Set<String>> lastNamesByCity

     *         = people.stream().collect(groupingBy(Person::getCity,

     *                                              mapping(Person::getLastName, toSet())));

     * }</pre>

     *

     * @param <T> the type of the input elements

     * @param <U> type of elements accepted by downstream collector

     * @param <A> intermediate accumulation type of the downstream collector

     * @param <R> result type of collector

     * @param mapper a function to be applied to the input elements

     * @param downstream a collector which will accept mapped values

     * @return a collector which applies the mapping function to the input

     * elements and provides the mapped results to the downstream collector

     */

    public static <T, U, A, R>

    Collector<T, ?, R> mapping(Function<? super T, ? extends U> mapper,

                               Collector<? super U, A, R> downstream) {

        BiConsumer<A, ? super U> downstreamAccumulator = downstream.accumulator();

        return new CollectorImpl<>(downstream.supplier(),

                                   (r, t) -> downstreamAccumulator.accept(r, mapper.apply(t)),

                                   downstream.combiner(), downstream.finisher(),

                                   downstream.characteristics());

    }

    /**

     * Adapts a {@code Collector} accepting elements of type {@code U} to one

     * accepting elements of type {@code T} by applying a flat mapping function

     * to each input element before accumulation.  The flat mapping function

     * maps an input element to a {@link Stream stream} covering zero or more

     * output elements that are then accumulated downstream.  Each mapped stream

     * is {@link java.util.stream.BaseStream#close() closed} after its contents

     * have been placed downstream.  (If a mapped stream is {@code null}

     * an empty stream is used, instead.)

     *

     * @apiNote

     * The {@code flatMapping()} collectors are most useful when used in a

     * multi-level reduction, such as downstream of a {@code groupingBy} or

     * {@code partitioningBy}.  For example, given a stream of

     * {@code Order}, to accumulate the set of line items for each customer:

     * <pre>{@code

     *     Map<String, Set<LineItem>> itemsByCustomerName

     *         = orders.stream().collect(groupingBy(Order::getCustomerName,

     *                                              flatMapping(order -> order.getLineItems().stream(), toSet())));

     * }</pre>

     *

     * @param <T> the type of the input elements

     * @param <U> type of elements accepted by downstream collector

     * @param <A> intermediate accumulation type of the downstream collector

     * @param <R> result type of collector

     * @param mapper a function to be applied to the input elements, which

     * returns a stream of results

     * @param downstream a collector which will receive the elements of the

     * stream returned by mapper

     * @return a collector which applies the mapping function to the input

     * elements and provides the flat mapped results to the downstream collector

     * @since 1.9

     */

    public static <T, U, A, R>

    Collector<T, ?, R> flatMapping(Function<? super T, ? extends Stream<? extends U>> mapper,

                                   Collector<? super U, A, R> downstream) {

        BiConsumer<A, ? super U> downstreamAccumulator = downstream.accumulator();

        return new CollectorImpl<>(downstream.supplier(),

                            (r, t) -> {

                                try (Stream<? extends U> result = mapper.apply(t)) {

                                    if (result != null)

                                        result.sequential().forEach(u -> downstreamAccumulator.accept(r, u));

                                }

                            },

                            downstream.combiner(), downstream.finisher(),

                            downstream.characteristics());

    }

    /**

     * Adapts a {@code Collector} to perform an additional finishing

     * transformation.  For example, one could adapt the {@link #toList()}

     * collector to always produce an immutable list with:

     * <pre>{@code

     *     List<String> people

     *         = people.stream().collect(collectingAndThen(toList(), Collections::unmodifiableList));

     * }</pre>

     *

     * @param <T> the type of the input elements

     * @param <A> intermediate accumulation type of the downstream collector

     * @param <R> result type of the downstream collector

     * @param <RR> result type of the resulting collector

     * @param downstream a collector

     * @param finisher a function to be applied to the final result of the downstream collector

     * @return a collector which performs the action of the downstream collector,

     * followed by an additional finishing step

     */

    public static<T,A,R,RR> Collector<T,A,RR> collectingAndThen(Collector<T,A,R> downstream,

                                                                Function<R,RR> finisher) {

        Set<Collector.Characteristics> characteristics = downstream.characteristics();

        if (characteristics.contains(Collector.Characteristics.IDENTITY_FINISH)) {

            if (characteristics.size() == 1)

                characteristics = Collectors.CH_NOID;

            else {

                characteristics = EnumSet.copyOf(characteristics);

                characteristics.remove(Collector.Characteristics.IDENTITY_FINISH);

                characteristics = Collections.unmodifiableSet(characteristics);

            }

        }

        return new CollectorImpl<>(downstream.supplier(),

                                   downstream.accumulator(),

                                   downstream.combiner(),

                                   downstream.finisher().andThen(finisher),

                                   characteristics);

    }

    /**

     * Returns a {@code Collector} accepting elements of type {@code T} that

     * counts the number of input elements.  If no elements are present, the

     * result is 0.

     *

     * @implSpec

     * This produces a result equivalent to:

     * <pre>{@code

     *     reducing(0L, e -> 1L, Long::sum)

     * }</pre>

     *

     * @param <T> the type of the input elements

     * @return a {@code Collector} that counts the input elements

     */