/*

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

import java.util.Objects;

import java.util.Spliterator;

import java.util.concurrent.CountedCompleter;

import java.util.concurrent.atomic.AtomicBoolean;

import java.util.function.Consumer;

import java.util.function.DoubleConsumer;

import java.util.function.DoublePredicate;

import java.util.function.IntConsumer;

import java.util.function.IntFunction;

import java.util.function.IntPredicate;

import java.util.function.LongConsumer;

import java.util.function.LongPredicate;

import java.util.function.Predicate;

/**

 * Factory for instances of a takeWhile and dropWhile operations

 * that produce subsequences of their input stream.

 *

 * @since 9

 */

final class WhileOps {

    static final int TAKE_FLAGS = StreamOpFlag.NOT_SIZED | StreamOpFlag.IS_SHORT_CIRCUIT;

    static final int DROP_FLAGS = StreamOpFlag.NOT_SIZED;

    /**

     * Appends a "takeWhile" operation to the provided Stream.

     *

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

     * @param upstream a reference stream with element type T

     * @param predicate the predicate that returns false to halt taking.

     */

    static <T> Stream<T> makeTakeWhileRef(AbstractPipeline<?, T, ?> upstream,

                                          Predicate<? super T> predicate) {

        Objects.requireNonNull(predicate);

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

            @Override

            <P_IN> Spliterator<T> opEvaluateParallelLazy(PipelineHelper<T> helper,

                                                         Spliterator<P_IN> spliterator) {

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

                    return opEvaluateParallel(helper, spliterator, Nodes.castingArray())

                            .spliterator();

                }

                else {

                    return new UnorderedWhileSpliterator.OfRef.Taking<>(

                            helper.wrapSpliterator(spliterator), false, predicate);

                }

            }

            @Override

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

                                              Spliterator<P_IN> spliterator,

                                              IntFunction<T[]> generator) {

                return new TakeWhileTask<>(this, helper, spliterator, generator)

                        .invoke();

            }

            @Override

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

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

                    boolean take = true;

                    @Override

                    public void begin(long size) {

                        downstream.begin(-1);

                    }

                    @Override

                    public void accept(T t) {

                            downstream.accept(t);

                        }

                    }

                    @Override

                    public boolean cancellationRequested() {

                        return !take || downstream.cancellationRequested();

                    }

                };

            }

        };

    }

    /**

     * Appends a "takeWhile" operation to the provided IntStream.

     *

     * @param upstream a reference stream with element type T

     * @param predicate the predicate that returns false to halt taking.

     */

    static IntStream makeTakeWhileInt(AbstractPipeline<?, Integer, ?> upstream,

                                      IntPredicate predicate) {

        Objects.requireNonNull(predicate);

        return new IntPipeline.StatefulOp<Integer>(upstream, StreamShape.INT_VALUE, TAKE_FLAGS) {

            @Override

            <P_IN> Spliterator<Integer> opEvaluateParallelLazy(PipelineHelper<Integer> helper,

                                                               Spliterator<P_IN> spliterator) {

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

                    return opEvaluateParallel(helper, spliterator, Integer[]::new)

                            .spliterator();

                }

                else {

                    return new UnorderedWhileSpliterator.OfInt.Taking(

                            (Spliterator.OfInt) helper.wrapSpliterator(spliterator), false, predicate);

                }

            }

            @Override

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

                                                    Spliterator<P_IN> spliterator,

                                                    IntFunction<Integer[]> generator) {

                return new TakeWhileTask<>(this, helper, spliterator, generator)

                        .invoke();

            }

            @Override

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

                return new Sink.ChainedInt<Integer>(sink) {

                    boolean take = true;

                    @Override

                    public void begin(long size) {

                        downstream.begin(-1);

                    }

                    @Override

                    public void accept(int t) {

                            downstream.accept(t);

                        }

                    }

                    @Override

                    public boolean cancellationRequested() {

                        return !take || downstream.cancellationRequested();

                    }

                };

            }

        };

    }

    /**

     * Appends a "takeWhile" operation to the provided LongStream.

     *

     * @param upstream a reference stream with element type T

     * @param predicate the predicate that returns false to halt taking.

     */

    static LongStream makeTakeWhileLong(AbstractPipeline<?, Long, ?> upstream,

                                        LongPredicate predicate) {

        Objects.requireNonNull(predicate);

        return new LongPipeline.StatefulOp<Long>(upstream, StreamShape.LONG_VALUE, TAKE_FLAGS) {

            @Override

            <P_IN> Spliterator<Long> opEvaluateParallelLazy(PipelineHelper<Long> helper,

                                                            Spliterator<P_IN> spliterator) {

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

                    return opEvaluateParallel(helper, spliterator, Long[]::new)

                            .spliterator();

                }

                else {

                    return new UnorderedWhileSpliterator.OfLong.Taking(

                            (Spliterator.OfLong) helper.wrapSpliterator(spliterator), false, predicate);

                }

            }

            @Override

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

                                                 Spliterator<P_IN> spliterator,

                                                 IntFunction<Long[]> generator) {

                return new TakeWhileTask<>(this, helper, spliterator, generator)

                        .invoke();

            }

            @Override

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

                return new Sink.ChainedLong<Long>(sink) {

                    boolean take = true;

                    @Override

                    public void begin(long size) {

                        downstream.begin(-1);

                    }

                    @Override

                    public void accept(long t) {

                            downstream.accept(t);

                        }

                    }

                    @Override

                    public boolean cancellationRequested() {

                        return !take || downstream.cancellationRequested();

                    }

                };

            }

        };

    }

    /**

     * Appends a "takeWhile" operation to the provided DoubleStream.

     *

     * @param upstream a reference stream with element type T

     * @param predicate the predicate that returns false to halt taking.

     */

    static DoubleStream makeTakeWhileDouble(AbstractPipeline<?, Double, ?> upstream,

                                            DoublePredicate predicate) {

        Objects.requireNonNull(predicate);

        return new DoublePipeline.StatefulOp<Double>(upstream, StreamShape.DOUBLE_VALUE, TAKE_FLAGS) {

            @Override

            <P_IN> Spliterator<Double> opEvaluateParallelLazy(PipelineHelper<Double> helper,

                                                              Spliterator<P_IN> spliterator) {

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

                    return opEvaluateParallel(helper, spliterator, Double[]::new)

                            .spliterator();

                }

                else {

                    return new UnorderedWhileSpliterator.OfDouble.Taking(

                            (Spliterator.OfDouble) helper.wrapSpliterator(spliterator), false, predicate);

                }

            }

            @Override

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

                                                   Spliterator<P_IN> spliterator,

                                                   IntFunction<Double[]> generator) {

                return new TakeWhileTask<>(this, helper, spliterator, generator)

                        .invoke();

            }

            @Override

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

                return new Sink.ChainedDouble<Double>(sink) {

                    boolean take = true;

                    @Override

                    public void begin(long size) {

                        downstream.begin(-1);

                    }

                    @Override

                    public void accept(double t) {

                            downstream.accept(t);

                        }

                    }

                    @Override

                    public boolean cancellationRequested() {

                        return !take || downstream.cancellationRequested();

                    }

                };

            }

        };

    }

    /**

     * A specialization for the dropWhile operation that controls if

     * elements to be dropped are counted and passed downstream.

     * <p>

     * This specialization is utilized by the {@link TakeWhileTask} for

     * pipelines that are ordered.  In such cases elements cannot be dropped

     * until all elements have been collected.

     *

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

     */

    interface DropWhileOp<T> {

        /**

         * Accepts a {@code Sink} which will receive the results of this

         * dropWhile operation, and return a {@code DropWhileSink} which

         * accepts

         * elements and which performs the dropWhile operation passing the

         * results to the provided {@code Sink}.

         *

         * @param sink sink to which elements should be sent after processing

         * @param retainAndCountDroppedElements true if elements to be dropped

         * are counted and passed to the sink, otherwise such elements

         * are actually dropped and not passed to the sink.

         * @return a dropWhile sink

         */

        DropWhileSink<T> opWrapSink(Sink<T> sink, boolean retainAndCountDroppedElements);

    }

    /**

     * A specialization for a dropWhile sink.

     *

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

     */

    interface DropWhileSink<T> extends Sink<T> {

        /**

         * @return the could of elements that would have been dropped and

         * instead were passed downstream.

         */

        long getDropCount();

    }

    /**

     * Appends a "dropWhile" operation to the provided Stream.

     *

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

     * @param upstream a reference stream with element type T

     * @param predicate the predicate that returns false to halt dropping.

     */

    static <T> Stream<T> makeDropWhileRef(AbstractPipeline<?, T, ?> upstream,

                                          Predicate<? super T> predicate) {

        Objects.requireNonNull(predicate);

        class Op extends ReferencePipeline.StatefulOp<T, T> implements DropWhileOp<T> {

            public Op(AbstractPipeline<?, T, ?> upstream, StreamShape inputShape, int opFlags) {

                super(upstream, inputShape, opFlags);

            }

            @Override

            <P_IN> Spliterator<T> opEvaluateParallelLazy(PipelineHelper<T> helper,

                                                         Spliterator<P_IN> spliterator) {

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

                    return opEvaluateParallel(helper, spliterator, Nodes.castingArray())

                            .spliterator();

                }

                else {

                    return new UnorderedWhileSpliterator.OfRef.Dropping<>(

                            helper.wrapSpliterator(spliterator), false, predicate);

                }

            }

            @Override

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

                                              Spliterator<P_IN> spliterator,

                                              IntFunction<T[]> generator) {

                return new DropWhileTask<>(this, helper, spliterator, generator)

                        .invoke();

            }

            @Override

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

                return opWrapSink(sink, false);

            }

            public DropWhileSink<T> opWrapSink(Sink<T> sink, boolean retainAndCountDroppedElements) {

                class OpSink extends Sink.ChainedReference<T, T> implements DropWhileSink<T> {

                    long dropCount;

                    boolean take;

                    OpSink() {

                        super(sink);

                    }

                    @Override

                    public void accept(T t) {

                        boolean takeElement = take || (take = !predicate.test(t));

                        // If ordered and element is dropped increment index

                        // for possible future truncation

                        if (retainAndCountDroppedElements && !takeElement)

                            dropCount++;

                        // If ordered need to process element, otherwise

                        // skip if element is dropped

                        if (retainAndCountDroppedElements || takeElement)

                            downstream.accept(t);

                    }

                    @Override

                    public long getDropCount() {

                        return dropCount;

                    }

                }

                return new OpSink();

            }

        }

        return new Op(upstream, StreamShape.REFERENCE, DROP_FLAGS);

    }

    /**

     * Appends a "dropWhile" operation to the provided IntStream.

     *

     * @param upstream a reference stream with element type T

     * @param predicate the predicate that returns false to halt dropping.

     */

    static IntStream makeDropWhileInt(AbstractPipeline<?, Integer, ?> upstream,

                                      IntPredicate predicate) {

        Objects.requireNonNull(predicate);

        class Op extends IntPipeline.StatefulOp<Integer> implements DropWhileOp<Integer> {

            public Op(AbstractPipeline<?, Integer, ?> upstream, StreamShape inputShape, int opFlags) {

                super(upstream, inputShape, opFlags);

            }

            @Override

            <P_IN> Spliterator<Integer> opEvaluateParallelLazy(PipelineHelper<Integer> helper,

                                                               Spliterator<P_IN> spliterator) {

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

                    return opEvaluateParallel(helper, spliterator, Integer[]::new)

                            .spliterator();

                }

                else {

                    return new UnorderedWhileSpliterator.OfInt.Dropping(

                            (Spliterator.OfInt) helper.wrapSpliterator(spliterator), false, predicate);

                }

            }

            @Override

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

                                                    Spliterator<P_IN> spliterator,

                                                    IntFunction<Integer[]> generator) {

                return new DropWhileTask<>(this, helper, spliterator, generator)

                        .invoke();

            }

            @Override

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

                return opWrapSink(sink, false);

            }

            public DropWhileSink<Integer> opWrapSink(Sink<Integer> sink, boolean retainAndCountDroppedElements) {

                class OpSink extends Sink.ChainedInt<Integer> implements DropWhileSink<Integer> {

                    long dropCount;

                    boolean take;

                    OpSink() {

                        super(sink);

                    }

                    @Override

                    public void accept(int t) {

                        boolean takeElement = take || (take = !predicate.test(t));

                        // If ordered and element is dropped increment index

                        // for possible future truncation

                        if (retainAndCountDroppedElements && !takeElement)

                            dropCount++;

                        // If ordered need to process element, otherwise

                        // skip if element is dropped

                        if (retainAndCountDroppedElements || takeElement)

                            downstream.accept(t);

                    }

                    @Override

                    public long getDropCount() {

                        return dropCount;

                    }

                }

                return new OpSink();

            }

        }

        return new Op(upstream, StreamShape.INT_VALUE, DROP_FLAGS);

    }

    /**

     * Appends a "dropWhile" operation to the provided LongStream.

     *

     * @param upstream a reference stream with element type T

     * @param predicate the predicate that returns false to halt dropping.

     */

    static LongStream makeDropWhileLong(AbstractPipeline<?, Long, ?> upstream,

                                        LongPredicate predicate) {

        Objects.requireNonNull(predicate);

        class Op extends LongPipeline.StatefulOp<Long> implements DropWhileOp<Long> {

            public Op(AbstractPipeline<?, Long, ?> upstream, StreamShape inputShape, int opFlags) {

                super(upstream, inputShape, opFlags);

            }

            @Override

            <P_IN> Spliterator<Long> opEvaluateParallelLazy(PipelineHelper<Long> helper,

                                                            Spliterator<P_IN> spliterator) {

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

                    return opEvaluateParallel(helper, spliterator, Long[]::new)

                            .spliterator();

                }

                else {

                    return new UnorderedWhileSpliterator.OfLong.Dropping(

                            (Spliterator.OfLong) helper.wrapSpliterator(spliterator), false, predicate);

                }

            }

            @Override

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

                                                 Spliterator<P_IN> spliterator,

                                                 IntFunction<Long[]> generator) {

                return new DropWhileTask<>(this, helper, spliterator, generator)

                        .invoke();

            }

            @Override

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

                return opWrapSink(sink, false);

            }

            public DropWhileSink<Long> opWrapSink(Sink<Long> sink, boolean retainAndCountDroppedElements) {

                class OpSink extends Sink.ChainedLong<Long> implements DropWhileSink<Long> {

                    long dropCount;

                    boolean take;

                    OpSink() {

                        super(sink);

                    }

                    @Override

                    public void accept(long t) {

                        boolean takeElement = take || (take = !predicate.test(t));