/*

 * 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.HashSet;

import java.util.LinkedHashSet;

import java.util.Objects;

import java.util.Set;

import java.util.Spliterator;

import java.util.concurrent.ConcurrentHashMap;

import java.util.concurrent.atomic.AtomicBoolean;

import java.util.function.IntFunction;

/**

 * Factory methods for transforming streams into duplicate-free streams, using

 * {@link Object#equals(Object)} to determine equality.

 *

 * @since 1.8

 */

final class DistinctOps {

    private DistinctOps() { }

    /**

     * Appends a "distinct" operation to the provided stream, and returns the

     * new stream.

     *

     * @param <T> the type of both input and output elements

     * @param upstream a reference stream with element type T

     * @return the new stream

     */

    static <T> ReferencePipeline<T, T> makeRef(AbstractPipeline<?, T, ?> upstream) {

        return new ReferencePipeline.StatefulOp<T, T>(upstream, StreamShape.REFERENCE,

                                                      StreamOpFlag.IS_DISTINCT | StreamOpFlag.NOT_SIZED) {

            <P_IN> Node<T> reduce(PipelineHelper<T> helper, Spliterator<P_IN> spliterator) {

                // If the stream is SORTED then it should also be ORDERED so the following will also

                // preserve the sort order

                TerminalOp<T, LinkedHashSet<T>> reduceOp

                        = ReduceOps.<T, LinkedHashSet<T>>makeRef(LinkedHashSet::new, LinkedHashSet::add,

                                                                 LinkedHashSet::addAll);

                return Nodes.node(reduceOp.evaluateParallel(helper, spliterator));

            }

            @Override

            <P_IN> Node<T> opEvaluateParallel(PipelineHelper<T> helper,

                                              Spliterator<P_IN> spliterator,

                                              IntFunction<T[]> generator) {

                if (StreamOpFlag.DISTINCT.isKnown(helper.getStreamAndOpFlags())) {

                    // No-op

                    return helper.evaluate(spliterator, false, generator);

                }

                else if (StreamOpFlag.ORDERED.isKnown(helper.getStreamAndOpFlags())) {

                    return reduce(helper, spliterator);

                }

                else {

                    // Holder of null state since ConcurrentHashMap does not support null values

                    AtomicBoolean seenNull = new AtomicBoolean(false);

                    ConcurrentHashMap<T, Boolean> map = new ConcurrentHashMap<>();

                    TerminalOp<T, Void> forEachOp = ForEachOps.makeRef(t -> {

                        if (t == null)

                            seenNull.set(true);

                        else

                            map.putIfAbsent(t, Boolean.TRUE);

                    }, false);

                    forEachOp.evaluateParallel(helper, spliterator);

                    // If null has been seen then copy the key set into a HashSet that supports null values

                    // and add null

                    Set<T> keys = map.keySet();

                    if (seenNull.get()) {

                        // TODO Implement a more efficient set-union view, rather than copying

                        keys = new HashSet<>(keys);

                        keys.add(null);

                    }

                    return Nodes.node(keys);

                }

            }

            @Override

            <P_IN> Spliterator<T> opEvaluateParallelLazy(PipelineHelper<T> helper, Spliterator<P_IN> spliterator) {

                if (StreamOpFlag.DISTINCT.isKnown(helper.getStreamAndOpFlags())) {

                    // No-op

                    return helper.wrapSpliterator(spliterator);

                }

                else if (StreamOpFlag.ORDERED.isKnown(helper.getStreamAndOpFlags())) {

                    // Not lazy, barrier required to preserve order

                    return reduce(helper, spliterator).spliterator();

                }

                else {

                    // Lazy

                    return new StreamSpliterators.DistinctSpliterator<>(helper.wrapSpliterator(spliterator));

                }

            }

            @Override

            Sink<T> opWrapSink(int flags, Sink<T> sink) {

                Objects.requireNonNull(sink);

                if (StreamOpFlag.DISTINCT.isKnown(flags)) {

                    return sink;

                } else if (StreamOpFlag.SORTED.isKnown(flags)) {

                    return new Sink.ChainedReference<T, T>(sink) {

                        boolean seenNull;

                        T lastSeen;

                        @Override

                        public void begin(long size) {

                            seenNull = false;

                            lastSeen = null;

                            downstream.begin(-1);

                        }

                        @Override

                        public void end() {

                            seenNull = false;

                            lastSeen = null;

                            downstream.end();

                        }

                        @Override

                        public void accept(T t) {

                            if (t == null) {

                                if (!seenNull) {

                                    seenNull = true;

                                    downstream.accept(lastSeen = null);

                                }

                            } else if (lastSeen == null || !t.equals(lastSeen)) {

                                downstream.accept(lastSeen = t);

                            }

                        }

                    };

                } else {

                    return new Sink.ChainedReference<T, T>(sink) {

                        Set<T> seen;

                        @Override

                        public void begin(long size) {

                            seen = new HashSet<>();

                            downstream.begin(-1);

                        }

                        @Override

                        public void end() {

                            seen = null;

                            downstream.end();

                        }

                        @Override

                        public void accept(T t) {

                                downstream.accept(t);

                            }

                        }

                    };

                }

            }

        };

    }

}