/*

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

import java.util.Arrays;

import java.util.Collection;

import java.util.Deque;

import java.util.List;

import java.util.Objects;

import java.util.Spliterator;

import java.util.Spliterators;

import java.util.concurrent.CountedCompleter;

import java.util.function.Consumer;

import java.util.function.DoubleConsumer;

import java.util.function.IntConsumer;

import java.util.function.IntFunction;

import java.util.function.LongConsumer;

/**

 * Factory methods for constructing implementations of {@link Node} and

 * {@link Node.Builder} and their primitive specializations.  Fork/Join tasks

 * for collecting output from a {@link PipelineHelper} to a {@link Node} and

 * flattening {@link Node}s.

 *

 * @since 1.8

 */

final class Nodes {

    private Nodes() {

        throw new Error("no instances");

    }

    /**

     * The maximum size of an array that can be allocated.

     */

    static final long MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;

    private static final Node EMPTY_NODE = new EmptyNode.OfRef();

    private static final Node.OfInt EMPTY_INT_NODE = new EmptyNode.OfInt();

    private static final Node.OfLong EMPTY_LONG_NODE = new EmptyNode.OfLong();

    private static final Node.OfDouble EMPTY_DOUBLE_NODE = new EmptyNode.OfDouble();

    // General shape-based node creation methods

    /**

     * Produces an empty node whose count is zero, has no children and no content.

     *

     * @param <T> the type of elements of the created node

     * @param shape the shape of the node to be created

     * @return an empty node.

     */

    @SuppressWarnings("unchecked")

    static <T> Node<T> emptyNode(StreamShape shape) {

        switch (shape) {

            case REFERENCE:    return (Node<T>) EMPTY_NODE;

            case INT_VALUE:    return (Node<T>) EMPTY_INT_NODE;

            case LONG_VALUE:   return (Node<T>) EMPTY_LONG_NODE;

            case DOUBLE_VALUE: return (Node<T>) EMPTY_DOUBLE_NODE;

            default:

                throw new IllegalStateException("Unknown shape " + shape);

        }

    }

    /**

     * Produces a concatenated {@link Node} that has two or more children.

     * <p>The count of the concatenated node is equal to the sum of the count

     * of each child. Traversal of the concatenated node traverses the content

     * of each child in encounter order of the list of children. Splitting a

     * spliterator obtained from the concatenated node preserves the encounter

     * order of the list of children.

     *

     * <p>The result may be a concatenated node, the input sole node if the size

     * of the list is 1, or an empty node.

     *

     * @param <T> the type of elements of the concatenated node

     * @param shape the shape of the concatenated node to be created

     * @param nodes the input nodes

     * @return a {@code Node} covering the elements of the input nodes

     * @throws IllegalStateException if all {@link Node} elements of the list

     * are an not instance of type supported by this factory.

     */

    @SuppressWarnings("unchecked")

    static <T> Node<T> conc(StreamShape shape, List<? extends Node<T>> nodes) {

        int size = nodes.size();

        if (size == 0)

            return emptyNode(shape);

        else if (size == 1)

            return nodes.get(0);

        else {

            // Create a right-balanced tree when there are more that 2 nodes

            switch (shape) {

                case REFERENCE: {

                    List<Node<T>> refNodes = (List<Node<T>>) nodes;

                    ConcNode<T> c = new ConcNode<>(refNodes.get(size - 2), refNodes.get(size - 1));

                    for (int i = size - 3; i >= 0; i--) {

                        c = new ConcNode<>(refNodes.get(i), c);

                    }

                    return c;

                }

                case INT_VALUE: {

                    List<? extends Node.OfInt> intNodes = (List<? extends Node.OfInt>) nodes;

                    IntConcNode c = new IntConcNode(intNodes.get(size - 2), intNodes.get(size - 1));

                    for (int i = size - 3; i >= 0; i--) {

                        c = new IntConcNode(intNodes.get(i), c);

                    }

                    return (Node<T>) c;

                }

                case LONG_VALUE: {

                    List<? extends Node.OfLong> longNodes = (List<? extends Node.OfLong>) nodes;

                    LongConcNode c = new LongConcNode(longNodes.get(size - 2), longNodes.get(size - 1));

                    for (int i = size - 3; i >= 0; i--) {

                        c = new LongConcNode(longNodes.get(i), c);

                    }

                    return (Node<T>) c;

                }

                case DOUBLE_VALUE: {

                    List<? extends Node.OfDouble> doubleNodes = (List<? extends Node.OfDouble>) nodes;

                    DoubleConcNode c = new DoubleConcNode(doubleNodes.get(size - 2), doubleNodes.get(size - 1));

                    for (int i = size - 3; i >= 0; i--) {

                        c = new DoubleConcNode(doubleNodes.get(i), c);

                    }

                    return (Node<T>) c;

                }

                default:

                    throw new IllegalStateException("Unknown shape " + shape);

            }

        }

    }

    /**

     * Truncate a {@link Node}, returning a node describing a subsequence of

     * the contents of the input node.

     *

     * @param <T> the type of elements of the input node and truncated node

     * @param input the input node

     * @param from the starting offset to include in the truncated node (inclusive)

     * @param to the ending offset ot include in the truncated node (exclusive)

     * @param generator the array factory (only used for reference nodes)

     * @return the truncated node

     */

    @SuppressWarnings("unchecked")

    static <T> Node<T> truncateNode(Node<T> input, long from, long to, IntFunction<T[]> generator) {

        StreamShape shape = input.getShape();

        long size = truncatedSize(input.count(), from, to);

        if (size == 0)

            return emptyNode(shape);

        else if (from == 0 && to >= input.count())

            return input;

        switch (shape) {

            case REFERENCE: {

                Spliterator<T> spliterator = input.spliterator();

                Node.Builder<T> nodeBuilder = Nodes.builder(size, generator);

                nodeBuilder.begin(size);

                for (int i = 0; i < from && spliterator.tryAdvance(e -> { }); i++) { }

                for (int i = 0; (i < size) && spliterator.tryAdvance(nodeBuilder); i++) { }

                nodeBuilder.end();

                return nodeBuilder.build();

            }

            case INT_VALUE: {

                Spliterator.OfInt spliterator = ((Node.OfInt) input).spliterator();

                Node.Builder.OfInt nodeBuilder = Nodes.intBuilder(size);

                nodeBuilder.begin(size);

                for (int i = 0; i < from && spliterator.tryAdvance((IntConsumer) e -> { }); i++) { }

                for (int i = 0; (i < size) && spliterator.tryAdvance((IntConsumer) nodeBuilder); i++) { }

                nodeBuilder.end();

                return (Node<T>) nodeBuilder.build();

            }

            case LONG_VALUE: {

                Spliterator.OfLong spliterator = ((Node.OfLong) input).spliterator();

                Node.Builder.OfLong nodeBuilder = Nodes.longBuilder(size);

                nodeBuilder.begin(size);

                for (int i = 0; i < from && spliterator.tryAdvance((LongConsumer) e -> { }); i++) { }

                for (int i = 0; (i < size) && spliterator.tryAdvance((LongConsumer) nodeBuilder); i++) { }

                nodeBuilder.end();

                return (Node<T>) nodeBuilder.build();

            }

            case DOUBLE_VALUE: {

                Spliterator.OfDouble spliterator = ((Node.OfDouble) input).spliterator();

                Node.Builder.OfDouble nodeBuilder = Nodes.doubleBuilder(size);

                nodeBuilder.begin(size);

                for (int i = 0; i < from && spliterator.tryAdvance((DoubleConsumer) e -> { }); i++) { }

                for (int i = 0; (i < size) && spliterator.tryAdvance((DoubleConsumer) nodeBuilder); i++) { }

                nodeBuilder.end();

                return (Node<T>) nodeBuilder.build();

            }

            default:

                throw new IllegalStateException("Unknown shape " + shape);

        }

    }

    private static long truncatedSize(long size, long from, long to) {

        if (from >= 0)

            size = Math.max(0, size - from);

        long limit = to - from;

        if (limit >= 0)

            size = Math.min(size, limit);

        return size;

    }

    // Reference-based node methods

    /**

     * Produces a {@link Node} describing an array.

     *

     * <p>The node will hold a reference to the array and will not make a copy.

     *

     * @param <T> the type of elements held by the node

     * @param array the array

     * @return a node holding an array

     */

    static <T> Node<T> node(T[] array) {

        return new ArrayNode<>(array);

    }

    /**

     * Produces a {@link Node} describing a {@link Collection}.

     * <p>

     * The node will hold a reference to the collection and will not make a copy.

     *

     * @param <T> the type of elements held by the node

     * @param c the collection

     * @return a node holding a collection

     */

    static <T> Node<T> node(Collection<T> c) {

        return new CollectionNode<>(c);

    }

    /**

     * Produces a {@link Node.Builder}.

     *

     * @param exactSizeIfKnown -1 if a variable size builder is requested,

     * otherwise the exact capacity desired.  A fixed capacity builder will

     * fail if the wrong number of elements are added to the builder.

     * @param generator the array factory

     * @param <T> the type of elements of the node builder

     * @return a {@code Node.Builder}

     */

    static <T> Node.Builder<T> builder(long exactSizeIfKnown, IntFunction<T[]> generator) {

        return (exactSizeIfKnown >= 0 && exactSizeIfKnown < MAX_ARRAY_SIZE)

               ? new FixedNodeBuilder<>(exactSizeIfKnown, generator)

               : builder();

    }

    /**

     * Produces a variable size @{link Node.Builder}.

     *

     * @param <T> the type of elements of the node builder

     * @return a {@code Node.Builder}

     */

    static <T> Node.Builder<T> builder() {

        return new SpinedNodeBuilder<>();

    }

    // Int nodes

    /**

     * Produces a {@link Node.OfInt} describing an int[] array.

     *

     * <p>The node will hold a reference to the array and will not make a copy.

     *

     * @param array the array

     * @return a node holding an array

     */

    static Node.OfInt node(int[] array) {

        return new IntArrayNode(array);

    }

    /**

     * Produces a {@link Node.Builder.OfInt}.

     *

     * @param exactSizeIfKnown -1 if a variable size builder is requested,

     * otherwise the exact capacity desired.  A fixed capacity builder will

     * fail if the wrong number of elements are added to the builder.

     * @return a {@code Node.Builder.OfInt}

     */

    static Node.Builder.OfInt intBuilder(long exactSizeIfKnown) {

        return (exactSizeIfKnown >= 0 && exactSizeIfKnown < MAX_ARRAY_SIZE)

               ? new IntFixedNodeBuilder(exactSizeIfKnown)

               : intBuilder();

    }

    /**

     * Produces a variable size @{link Node.Builder.OfInt}.

     *

     * @return a {@code Node.Builder.OfInt}

     */

    static Node.Builder.OfInt intBuilder() {

        return new IntSpinedNodeBuilder();

    }

    // Long nodes

    /**

     * Produces a {@link Node.OfLong} describing a long[] array.

     * <p>

     * The node will hold a reference to the array and will not make a copy.

     *

     * @param array the array

     * @return a node holding an array

     */

    static Node.OfLong node(final long[] array) {

        return new LongArrayNode(array);

    }

    /**

     * Produces a {@link Node.Builder.OfLong}.

     *

     * @param exactSizeIfKnown -1 if a variable size builder is requested,

     * otherwise the exact capacity desired.  A fixed capacity builder will

     * fail if the wrong number of elements are added to the builder.

     * @return a {@code Node.Builder.OfLong}

     */

    static Node.Builder.OfLong longBuilder(long exactSizeIfKnown) {

        return (exactSizeIfKnown >= 0 && exactSizeIfKnown < MAX_ARRAY_SIZE)

               ? new LongFixedNodeBuilder(exactSizeIfKnown)

               : longBuilder();

    }

    /**

     * Produces a variable size @{link Node.Builder.OfLong}.

     *

     * @return a {@code Node.Builder.OfLong}

     */

    static Node.Builder.OfLong longBuilder() {

        return new LongSpinedNodeBuilder();

    }

    // Double nodes

    /**

     * Produces a {@link Node.OfDouble} describing a double[] array.

     *

     * <p>The node will hold a reference to the array and will not make a copy.

     *

     * @param array the array

     * @return a node holding an array

     */

    static Node.OfDouble node(final double[] array) {

        return new DoubleArrayNode(array);

    }

    /**

     * Produces a {@link Node.Builder.OfDouble}.

     *

     * @param exactSizeIfKnown -1 if a variable size builder is requested,

     * otherwise the exact capacity desired.  A fixed capacity builder will

     * fail if the wrong number of elements are added to the builder.

     * @return a {@code Node.Builder.OfDouble}

     */

    static Node.Builder.OfDouble doubleBuilder(long exactSizeIfKnown) {

        return (exactSizeIfKnown >= 0 && exactSizeIfKnown < MAX_ARRAY_SIZE)

               ? new DoubleFixedNodeBuilder(exactSizeIfKnown)

               : doubleBuilder();

    }

    /**

     * Produces a variable size @{link Node.Builder.OfDouble}.

     *

     * @return a {@code Node.Builder.OfDouble}

     */

    static Node.Builder.OfDouble doubleBuilder() {

        return new DoubleSpinedNodeBuilder();

    }

    // Parallel evaluation of pipelines to nodes

    /**

     * Collect, in parallel, elements output from a pipeline and describe those

     * elements with a {@link Node}.

     *

     * @implSpec

     * If the exact size of the output from the pipeline is known and the source

     * {@link Spliterator} has the {@link Spliterator#SUBSIZED} characteristic,

     * then a flat {@link Node} will be returned whose content is an array,

     * since the size is known the array can be constructed in advance and

     * output elements can be placed into the array concurrently by leaf

     * tasks at the correct offsets.  If the exact size is not known, output

     * elements are collected into a conc-node whose shape mirrors that

     * of the computation. This conc-node can then be flattened in

     * parallel to produce a flat {@code Node} if desired.

     *

     * @param helper the pipeline helper describing the pipeline

     * @param flattenTree whether a conc node should be flattened into a node

     *                    describing an array before returning

     * @param generator the array generator

     * @return a {@link Node} describing the output elements

     */

    public static <P_IN, P_OUT> Node<P_OUT> collect(PipelineHelper<P_OUT> helper,

                                                    Spliterator<P_IN> spliterator,

                                                    boolean flattenTree,

                                                    IntFunction<P_OUT[]> generator) {

        long size = helper.exactOutputSizeIfKnown(spliterator);

        if (size >= 0 && spliterator.hasCharacteristics(Spliterator.SUBSIZED)) {

            if (size >= MAX_ARRAY_SIZE)

                throw new IllegalArgumentException("Stream size exceeds max array size");

            P_OUT[] array = generator.apply((int) size);

            new SizedCollectorTask.OfRef<>(spliterator, helper, array).invoke();

            return node(array);

        } else {

            Node<P_OUT> node = new CollectorTask<>(helper, generator, spliterator).invoke();

            return flattenTree ? flatten(node, generator) : node;

        }

    }

    /**

     * Collect, in parallel, elements output from an int-valued pipeline and

     * describe those elements with a {@link Node.OfInt}.

     *

     * @implSpec

     * If the exact size of the output from the pipeline is known and the source

     * {@link Spliterator} has the {@link Spliterator#SUBSIZED} characteristic,

     * then a flat {@link Node} will be returned whose content is an array,

     * since the size is known the array can be constructed in advance and

     * output elements can be placed into the array concurrently by leaf

     * tasks at the correct offsets.  If the exact size is not known, output

     * elements are collected into a conc-node whose shape mirrors that

     * of the computation. This conc-node can then be flattened in

     * parallel to produce a flat {@code Node.OfInt} if desired.

     *

     * @param <P_IN> the type of elements from the source Spliterator

     * @param helper the pipeline helper describing the pipeline

     * @param flattenTree whether a conc node should be flattened into a node

     *                    describing an array before returning

     * @return a {@link Node.OfInt} describing the output elements

     */

    public static <P_IN> Node.OfInt collectInt(PipelineHelper<Integer> helper,

                                               Spliterator<P_IN> spliterator,

                                               boolean flattenTree) {

        long size = helper.exactOutputSizeIfKnown(spliterator);

        if (size >= 0 && spliterator.hasCharacteristics(Spliterator.SUBSIZED)) {

            if (size >= MAX_ARRAY_SIZE)

                throw new IllegalArgumentException("Stream size exceeds max array size");

            int[] array = new int[(int) size];

            new SizedCollectorTask.OfInt<>(spliterator, helper, array).invoke();

            return node(array);

        }

        else {

            Node.OfInt node = new IntCollectorTask<>(helper, spliterator).invoke();

            return flattenTree ? flattenInt(node) : node;

        }

    }

    /**

     * Collect, in parallel, elements output from a long-valued pipeline and

     * describe those elements with a {@link Node.OfLong}.

     *

     * @implSpec

     * If the exact size of the output from the pipeline is known and the source

     * {@link Spliterator} has the {@link Spliterator#SUBSIZED} characteristic,

     * then a flat {@link Node} will be returned whose content is an array,

     * since the size is known the array can be constructed in advance and

     * output elements can be placed into the array concurrently by leaf

     * tasks at the correct offsets.  If the exact size is not known, output

     * elements are collected into a conc-node whose shape mirrors that

     * of the computation. This conc-node can then be flattened in

     * parallel to produce a flat {@code Node.OfLong} if desired.

     *

     * @param <P_IN> the type of elements from the source Spliterator

     * @param helper the pipeline helper describing the pipeline

     * @param flattenTree whether a conc node should be flattened into a node

     *                    describing an array before returning

     * @return a {@link Node.OfLong} describing the output elements

     */

    public static <P_IN> Node.OfLong collectLong(PipelineHelper<Long> helper,

                                                 Spliterator<P_IN> spliterator,

                                                 boolean flattenTree) {

        long size = helper.exactOutputSizeIfKnown(spliterator);

        if (size >= 0 && spliterator.hasCharacteristics(Spliterator.SUBSIZED)) {

            if (size >= MAX_ARRAY_SIZE)

                throw new IllegalArgumentException("Stream size exceeds max array size");

            long[] array = new long[(int) size];

            new SizedCollectorTask.OfLong<>(spliterator, helper, array).invoke();

            return node(array);

        }

        else {

            Node.OfLong node = new LongCollectorTask<>(helper, spliterator).invoke();

            return flattenTree ? flattenLong(node) : node;

        }

    }

    /**

     * Collect, in parallel, elements output from n double-valued pipeline and

     * describe those elements with a {@link Node.OfDouble}.

     *

     * @implSpec

     * If the exact size of the output from the pipeline is known and the source

     * {@link Spliterator} has the {@link Spliterator#SUBSIZED} characteristic,

     * then a flat {@link Node} will be returned whose content is an array,

     * since the size is known the array can be constructed in advance and

     * output elements can be placed into the array concurrently by leaf

     * tasks at the correct offsets.  If the exact size is not known, output

     * elements are collected into a conc-node whose shape mirrors that

     * of the computation. This conc-node can then be flattened in

     * parallel to produce a flat {@code Node.OfDouble} if desired.

     *

     * @param <P_IN> the type of elements from the source Spliterator

     * @param helper the pipeline helper describing the pipeline

     * @param flattenTree whether a conc node should be flattened into a node

     *                    describing an array before returning

     * @return a {@link Node.OfDouble} describing the output elements

     */

    public static <P_IN> Node.OfDouble collectDouble(PipelineHelper<Double> helper,

                                                     Spliterator<P_IN> spliterator,