/*

 * Copyright (c) 2013, 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.LongSummaryStatistics;

import java.util.Objects;

import java.util.OptionalDouble;

import java.util.OptionalLong;

import java.util.PrimitiveIterator;

import java.util.Spliterator;

import java.util.Spliterators;

import java.util.function.BiConsumer;

import java.util.function.BinaryOperator;

import java.util.function.IntFunction;

import java.util.function.LongBinaryOperator;

import java.util.function.LongConsumer;

import java.util.function.LongFunction;

import java.util.function.LongPredicate;

import java.util.function.LongToDoubleFunction;

import java.util.function.LongToIntFunction;

import java.util.function.LongUnaryOperator;

import java.util.function.ObjLongConsumer;

import java.util.function.Supplier;

/**

 * Abstract base class for an intermediate pipeline stage or pipeline source

 * stage implementing whose elements are of type {@code long}.

 *

 * @param <E_IN> type of elements in the upstream source

 * @since 1.8

 */

abstract class LongPipeline<E_IN>

        extends AbstractPipeline<E_IN, Long, LongStream>

        implements LongStream {

    /**

     * Constructor for the head of a stream pipeline.

     *

     * @param source {@code Supplier<Spliterator>} describing the stream source

     * @param sourceFlags the source flags for the stream source, described in

     *        {@link StreamOpFlag}

     * @param parallel {@code true} if the pipeline is parallel

     */

    LongPipeline(Supplier<? extends Spliterator<Long>> source,

                 int sourceFlags, boolean parallel) {

        super(source, sourceFlags, parallel);

    }

    /**

     * Constructor for the head of a stream pipeline.

     *

     * @param source {@code Spliterator} describing the stream source

     * @param sourceFlags the source flags for the stream source, described in

     *        {@link StreamOpFlag}

     * @param parallel {@code true} if the pipeline is parallel

     */

    LongPipeline(Spliterator<Long> source,

                 int sourceFlags, boolean parallel) {

        super(source, sourceFlags, parallel);

    }

    /**

     * Constructor for appending an intermediate operation onto an existing pipeline.

     *

     * @param upstream the upstream element source.

     * @param opFlags the operation flags

     */

    LongPipeline(AbstractPipeline<?, E_IN, ?> upstream, int opFlags) {

        super(upstream, opFlags);

    }

    /**

     * Adapt a {@code Sink<Long> to an {@code LongConsumer}, ideally simply

     * by casting.

     */

    private static LongConsumer adapt(Sink<Long> sink) {

        if (sink instanceof LongConsumer) {

            return (LongConsumer) sink;

        } else {

            if (Tripwire.ENABLED)

                Tripwire.trip(AbstractPipeline.class,

                              "using LongStream.adapt(Sink<Long> s)");

            return sink::accept;

        }

    }

    /**

     * Adapt a {@code Spliterator<Long>} to a {@code Spliterator.OfLong}.

     *

     * @implNote

     * The implementation attempts to cast to a Spliterator.OfLong, and throws

     * an exception if this cast is not possible.

     */

    private static Spliterator.OfLong adapt(Spliterator<Long> s) {

        if (s instanceof Spliterator.OfLong) {

            return (Spliterator.OfLong) s;

        } else {

            if (Tripwire.ENABLED)

                Tripwire.trip(AbstractPipeline.class,

                              "using LongStream.adapt(Spliterator<Long> s)");

            throw new UnsupportedOperationException("LongStream.adapt(Spliterator<Long> s)");

        }

    }

    // Shape-specific methods

    @Override

    final StreamShape getOutputShape() {

        return StreamShape.LONG_VALUE;

    }

    @Override

    final <P_IN> Node<Long> evaluateToNode(PipelineHelper<Long> helper,

                                           Spliterator<P_IN> spliterator,

                                           boolean flattenTree,

                                           IntFunction<Long[]> generator) {

        return Nodes.collectLong(helper, spliterator, flattenTree);

    }

    @Override

    final <P_IN> Spliterator<Long> wrap(PipelineHelper<Long> ph,

                                        Supplier<Spliterator<P_IN>> supplier,

                                        boolean isParallel) {

        return new StreamSpliterators.LongWrappingSpliterator<>(ph, supplier, isParallel);

    }

    @Override

    final Spliterator.OfLong lazySpliterator(Supplier<? extends Spliterator<Long>> supplier) {

        return new StreamSpliterators.DelegatingSpliterator.OfLong((Supplier<Spliterator.OfLong>) supplier);

    }

    @Override

    final void forEachWithCancel(Spliterator<Long> spliterator, Sink<Long> sink) {

        Spliterator.OfLong spl = adapt(spliterator);

        LongConsumer adaptedSink =  adapt(sink);

        do { } while (!sink.cancellationRequested() && spl.tryAdvance(adaptedSink));

    }

    @Override

    final Node.Builder<Long> makeNodeBuilder(long exactSizeIfKnown, IntFunction<Long[]> generator) {

        return Nodes.longBuilder(exactSizeIfKnown);

    }

    // LongStream

    @Override

    public final PrimitiveIterator.OfLong iterator() {

        return Spliterators.iterator(spliterator());

    }

    @Override

    public final Spliterator.OfLong spliterator() {

        return adapt(super.spliterator());

    }

    // Stateless intermediate ops from LongStream

    @Override

    public final DoubleStream asDoubleStream() {

        return new DoublePipeline.StatelessOp<Long>(this, StreamShape.LONG_VALUE,

                                                    StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT) {

            @Override

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

                return new Sink.ChainedLong(sink) {

                    @Override

                    public void accept(long t) {

                        downstream.accept((double) t);

                    }

                };

            }

        };

    }

    @Override

    public final Stream<Long> boxed() {

        return mapToObj(Long::valueOf);

    }

    @Override

    public final LongStream map(LongUnaryOperator mapper) {

        Objects.requireNonNull(mapper);

        return new StatelessOp<Long>(this, StreamShape.LONG_VALUE,

                                     StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT) {

            @Override

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

                return new Sink.ChainedLong(sink) {

                    @Override

                    public void accept(long t) {

                        downstream.accept(mapper.applyAsLong(t));

                    }

                };

            }

        };

    }

    @Override

    public final <U> Stream<U> mapToObj(LongFunction<? extends U> mapper) {

        Objects.requireNonNull(mapper);

        return new ReferencePipeline.StatelessOp<Long, U>(this, StreamShape.LONG_VALUE,

                                                          StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT) {

            @Override

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

                return new Sink.ChainedLong(sink) {

                    @Override

                    public void accept(long t) {

                        downstream.accept(mapper.apply(t));

                    }

                };

            }

        };

    }

    @Override

    public final IntStream mapToInt(LongToIntFunction mapper) {

        Objects.requireNonNull(mapper);

        return new IntPipeline.StatelessOp<Long>(this, StreamShape.LONG_VALUE,

                                                 StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT) {

            @Override

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

                return new Sink.ChainedLong(sink) {

                    @Override

                    public void accept(long t) {

                        downstream.accept(mapper.applyAsInt(t));

                    }

                };

            }

        };

    }

    @Override

    public final DoubleStream mapToDouble(LongToDoubleFunction mapper) {

        Objects.requireNonNull(mapper);

        return new DoublePipeline.StatelessOp<Long>(this, StreamShape.LONG_VALUE,

                                                    StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT) {

            @Override

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

                return new Sink.ChainedLong(sink) {

                    @Override

                    public void accept(long t) {

                        downstream.accept(mapper.applyAsDouble(t));

                    }

                };

            }

        };

    }

    @Override

    public final LongStream flatMap(LongFunction<? extends LongStream> mapper) {

        return new StatelessOp<Long>(this, StreamShape.LONG_VALUE,

                                     StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT | StreamOpFlag.NOT_SIZED) {

            @Override

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

                return new Sink.ChainedLong(sink) {

                    @Override

                    public void begin(long size) {

                        downstream.begin(-1);

                    }

                    @Override

                    public void accept(long t) {

                        // We can do better that this too; optimize for depth=0 case and just grab spliterator and forEach it

                        LongStream result = mapper.apply(t);

                        if (result != null)

                            result.sequential().forEach(i -> downstream.accept(i));

                    }

                };

            }

        };

    }

    @Override

    public LongStream unordered() {

        if (!isOrdered())

            return this;

        return new StatelessOp<Long>(this, StreamShape.LONG_VALUE, StreamOpFlag.NOT_ORDERED) {

            @Override

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

                return sink;

            }

        };

    }

    @Override

    public final LongStream filter(LongPredicate predicate) {

        Objects.requireNonNull(predicate);

        return new StatelessOp<Long>(this, StreamShape.LONG_VALUE,

                                     StreamOpFlag.NOT_SIZED) {

            @Override

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

                return new Sink.ChainedLong(sink) {

                    @Override

                    public void begin(long size) {

                        downstream.begin(-1);

                    }

                    @Override

                    public void accept(long t) {

                        if (predicate.test(t))

                            downstream.accept(t);

                    }

                };

            }

        };

    }

    @Override

    public final LongStream peek(LongConsumer consumer) {

        Objects.requireNonNull(consumer);

        return new StatelessOp<Long>(this, StreamShape.LONG_VALUE,

                                     0) {

            @Override

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

                return new Sink.ChainedLong(sink) {

                    @Override

                    public void accept(long t) {

                        consumer.accept(t);

                        downstream.accept(t);

                    }

                };

            }

        };

    }

    // Stateful intermediate ops from LongStream

    private LongStream slice(long skip, long limit) {

        return SliceOps.makeLong(this, skip, limit);

    }

    @Override

    public final LongStream limit(long maxSize) {

        if (maxSize < 0)

            throw new IllegalArgumentException(Long.toString(maxSize));

        return slice(0, maxSize);

    }

    @Override

    public final LongStream substream(long startingOffset) {

        if (startingOffset < 0)

            throw new IllegalArgumentException(Long.toString(startingOffset));

        if (startingOffset == 0)

            return this;

        else

            return slice(startingOffset, -1);

    }

    @Override

    public final LongStream substream(long startingOffset, long endingOffset) {

        if (startingOffset < 0 || endingOffset < startingOffset)

            throw new IllegalArgumentException(String.format("substream(%d, %d)", startingOffset, endingOffset));

        return slice(startingOffset, endingOffset - startingOffset);

    }

    @Override

    public final LongStream sorted() {

        return SortedOps.makeLong(this);

    }

    @Override

    public final LongStream distinct() {

        // While functional and quick to implement, this approach is not very efficient.

        // An efficient version requires a long-specific map/set implementation.

        return boxed().distinct().mapToLong(i -> (long) i);

    }

    // Terminal ops from LongStream

    @Override

    public void forEach(LongConsumer action) {

        evaluate(ForEachOps.makeLong(action, false));

    }

    @Override

    public void forEachOrdered(LongConsumer action) {

        evaluate(ForEachOps.makeLong(action, true));

    }

    @Override

    public final long sum() {

        // use better algorithm to compensate for intermediate overflow?

        return reduce(0, Long::sum);

    }

    @Override

    public final OptionalLong min() {

        return reduce(Math::min);

    }

    @Override

    public final OptionalLong max() {

        return reduce(Math::max);

    }

    @Override

    public final OptionalDouble average() {

        long[] avg = collect(() -> new long[2],

                             (ll, i) -> {

                                 ll[0]++;

                                 ll[1] += i;

                             },

                             (ll, rr) -> {

                                 ll[0] += rr[0];

                                 ll[1] += rr[1];

                             });

        return avg[0] > 0

               ? OptionalDouble.of((double) avg[1] / avg[0])

               : OptionalDouble.empty();

    }

    @Override

    public final long count() {

        return map(e -> 1L).sum();

    }

    @Override

    public final LongSummaryStatistics summaryStatistics() {

        return collect(LongSummaryStatistics::new, LongSummaryStatistics::accept,

                       LongSummaryStatistics::combine);

    }

    @Override

    public final long reduce(long identity, LongBinaryOperator op) {

        return evaluate(ReduceOps.makeLong(identity, op));

    }

    @Override

    public final OptionalLong reduce(LongBinaryOperator op) {

        return evaluate(ReduceOps.makeLong(op));

    }

    @Override

    public final <R> R collect(Supplier<R> resultFactory,

                               ObjLongConsumer<R> accumulator,

                               BiConsumer<R, R> combiner) {

        BinaryOperator<R> operator = (left, right) -> {

            combiner.accept(left, right);

            return left;

        };

        return evaluate(ReduceOps.makeLong(resultFactory, accumulator, operator));

    }

    @Override

    public final boolean anyMatch(LongPredicate predicate) {

        return evaluate(MatchOps.makeLong(predicate, MatchOps.MatchKind.ANY));

    }

    @Override

    public final boolean allMatch(LongPredicate predicate) {

        return evaluate(MatchOps.makeLong(predicate, MatchOps.MatchKind.ALL));

    }

    @Override

    public final boolean noneMatch(LongPredicate predicate) {

        return evaluate(MatchOps.makeLong(predicate, MatchOps.MatchKind.NONE));

    }

    @Override

    public final OptionalLong findFirst() {

        return evaluate(FindOps.makeLong(true));

    }

    @Override

    public final OptionalLong findAny() {

        return evaluate(FindOps.makeLong(false));

    }

    @Override

    public final long[] toArray() {

        return Nodes.flattenLong((Node.OfLong) evaluateToArrayNode(Long[]::new))

                .asPrimitiveArray();

    }

    //

    /**

     * Source stage of a LongPipeline.

     *

     * @param <E_IN> type of elements in the upstream source

     * @since 1.8

     */

    static class Head<E_IN> extends LongPipeline<E_IN> {

        /**

         * Constructor for the source stage of a LongStream.

         *

         * @param source {@code Supplier<Spliterator>} describing the stream

         *               source

         * @param sourceFlags the source flags for the stream source, described

         *                    in {@link StreamOpFlag}

         * @param parallel {@code true} if the pipeline is parallel

         */

        Head(Supplier<? extends Spliterator<Long>> source,

             int sourceFlags, boolean parallel) {

            super(source, sourceFlags, parallel);

        }

        /**

         * Constructor for the source stage of a LongStream.

         *

         * @param source {@code Spliterator} describing the stream source

         * @param sourceFlags the source flags for the stream source, described

         *                    in {@link StreamOpFlag}

         * @param parallel {@code true} if the pipeline is parallel