/*

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

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;

import java.util.function.LongFunction;

/**

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

    // IllegalArgumentException messages

    static final String BAD_SIZE = "Stream size exceeds max array size";

    @SuppressWarnings("rawtypes")

    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 left the left input node

     * @param right the right input node

     * @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, Node<T> left, Node<T> right) {

        switch (shape) {

            case REFERENCE:

                return new ConcNode<>(left, right);

            case INT_VALUE:

                return (Node<T>) new ConcNode.OfInt((Node.OfInt) left, (Node.OfInt) right);

            case LONG_VALUE:

                return (Node<T>) new ConcNode.OfLong((Node.OfLong) left, (Node.OfLong) right);

            case DOUBLE_VALUE:

                return (Node<T>) new ConcNode.OfDouble((Node.OfDouble) left, (Node.OfDouble) right);

            default:

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

        }

    }

    // 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(BAD_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.OfRef<>(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(BAD_SIZE);

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

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

            return node(array);

        }

        else {

            Node.OfInt node = new CollectorTask.OfInt<>(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(BAD_SIZE);

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

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

            return node(array);

        }

        else {

            Node.OfLong node = new CollectorTask.OfLong<>(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,

                                                     boolean flattenTree) {

        long size = helper.exactOutputSizeIfKnown(spliterator);

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

            if (size >= MAX_ARRAY_SIZE)

                throw new IllegalArgumentException(BAD_SIZE);

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

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

            return node(array);

        }

        else {

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

            return flattenTree ? flattenDouble(node) : node;

        }

    }

    // Parallel flattening of nodes

    /**

     * Flatten, in parallel, a {@link Node}.  A flattened node is one that has

     * no children.  If the node is already flat, it is simply returned.

     *

     * @implSpec

     * If a new node is to be created, the generator is used to create an array

     * whose length is {@link Node#count()}.  Then the node tree is traversed

     * and leaf node elements are placed in the array concurrently by leaf tasks

     * at the correct offsets.

     *

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

     * @param node the node to flatten

     * @param generator the array factory used to create array instances

     * @return a flat {@code Node}

     */

    public static <T> Node<T> flatten(Node<T> node, IntFunction<T[]> generator) {

        if (node.getChildCount() > 0) {

            long size = node.count();

            if (size >= MAX_ARRAY_SIZE)

                throw new IllegalArgumentException(BAD_SIZE);

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

            new ToArrayTask.OfRef<>(node, array, 0).invoke();

            return node(array);

        } else {

            return node;

        }

    }

    /**

     * Flatten, in parallel, a {@link Node.OfInt}.  A flattened node is one that

     * has no children.  If the node is already flat, it is simply returned.

     *

     * @implSpec

     * If a new node is to be created, a new int[] array is created whose length

     * is {@link Node#count()}.  Then the node tree is traversed and leaf node

     * elements are placed in the array concurrently by leaf tasks at the

     * correct offsets.

     *

     * @param node the node to flatten

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

     */

    public static Node.OfInt flattenInt(Node.OfInt node) {

        if (node.getChildCount() > 0) {

            long size = node.count();

            if (size >= MAX_ARRAY_SIZE)

                throw new IllegalArgumentException(BAD_SIZE);

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

            new ToArrayTask.OfInt(node, array, 0).invoke();

            return node(array);

        } else {

            return node;

        }

    }

    /**

     * Flatten, in parallel, a {@link Node.OfLong}.  A flattened node is one that

     * has no children.  If the node is already flat, it is simply returned.

     *

     * @implSpec

     * If a new node is to be created, a new long[] array is created whose length

     * is {@link Node#count()}.  Then the node tree is traversed and leaf node

     * elements are placed in the array concurrently by leaf tasks at the

     * correct offsets.

     *

     * @param node the node to flatten

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

     */

    public static Node.OfLong flattenLong(Node.OfLong node) {

        if (node.getChildCount() > 0) {

            long size = node.count();

            if (size >= MAX_ARRAY_SIZE)

                throw new IllegalArgumentException(BAD_SIZE);

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

            new ToArrayTask.OfLong(node, array, 0).invoke();

            return node(array);

        } else {