1 /*

     2  * Copyright (c) 2012, 2013, Oracle and/or its affiliates. All rights reserved.

     3  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.

     4  *

     5  * This code is free software; you can redistribute it and/or modify it

     6  * under the terms of the GNU General Public License version 2 only, as

     7  * published by the Free Software Foundation.  Oracle designates this

     8  * particular file as subject to the "Classpath" exception as provided

     9  * by Oracle in the LICENSE file that accompanied this code.

    10  *

    11  * This code is distributed in the hope that it will be useful, but WITHOUT

    12  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or

    13  * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License

    14  * version 2 for more details (a copy is included in the LICENSE file that

    15  * accompanied this code).

    16  *

    17  * You should have received a copy of the GNU General Public License version

    18  * 2 along with this work; if not, write to the Free Software Foundation,

    19  * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.

    20  *

    21  * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA

    22  * or visit www.oracle.com if you need additional information or have any

    23  * questions.

    24  */

    25 package java.util.stream;

    26 

    27 import java.util.Spliterator;

    28 import java.util.function.IntFunction;

    29 

    30 /**

    31  * Helper class for executing <a href="package-summary.html#StreamOps">

    32  * stream pipelines</a>, capturing all of the information about a stream

    33  * pipeline (output shape, intermediate operations, stream flags, parallelism,

    34  * etc) in one place.

    35  *

    36  * <p>

    37  * A {@code PipelineHelper} describes the initial segment of a stream pipeline,

    38  * including its source, intermediate operations, and may additionally

    39  * incorporate information about the terminal (or stateful) operation which

    40  * follows the last intermediate operation described by this

    41  * {@code PipelineHelper}. The {@code PipelineHelper} is passed to the

    42  * {@link TerminalOp#evaluateParallel(PipelineHelper, java.util.Spliterator)},

    43  * {@link TerminalOp#evaluateSequential(PipelineHelper, java.util.Spliterator)},

    44  * and {@link AbstractPipeline#opEvaluateParallel(PipelineHelper, java.util.Spliterator,

    45  * java.util.function.IntFunction)}, methods, which can use the

    46  * {@code PipelineHelper} to access information about the pipeline such as

    47  * head shape, stream flags, and size, and use the helper methods

    48  * such as {@link #wrapAndCopyInto(Sink, Spliterator)},

    49  * {@link #copyInto(Sink, Spliterator)}, and {@link #wrapSink(Sink)} to execute

    50  * pipeline operations.

    51  *

    52  * @param <P_OUT> type of output elements from the pipeline

    53  * @since 1.8

    54  */

    55 abstract class PipelineHelper<P_OUT> {

    56 

    57     /**

    58      * Gets the stream shape for the source of the pipeline segment.

    59      *

    60      * @return the stream shape for the source of the pipeline segment.

    61      */

    62     abstract StreamShape getSourceShape();

    63 

    64     /**

    65      * Gets the combined stream and operation flags for the output of the described

    66      * pipeline.  This will incorporate stream flags from the stream source, all

    67      * the intermediate operations and the terminal operation.

    68      *

    69      * @return the combined stream and operation flags

    70      * @see StreamOpFlag

    71      */

    72     abstract int getStreamAndOpFlags();

    73 

    74     /**

    75      * Returns the exact output size of the portion of the output resulting from

    76      * applying the pipeline stages described by this {@code PipelineHelper} to

    77      * the portion of the input described by the provided

    78      * {@code Spliterator}, if known.  If not known or known infinite, will

    79      * return {@code -1}.

    80      *

    81      * @apiNote

    82      * The exact output size is known if the {@code Spliterator} has the

    83      * {@code SIZED} characteristic, and the operation flags

    84      * {@link StreamOpFlag#SIZED} is known on the combined stream and operation

    85      * flags.

    86      *

    87      * @param spliterator the spliterator describing the relevant portion of the

    88      *        source data

    89      * @return the exact size if known, or -1 if infinite or unknown

    90      */

    91     abstract<P_IN> long exactOutputSizeIfKnown(Spliterator<P_IN> spliterator);

    92 

    93     /**

    94      * Applies the pipeline stages described by this {@code PipelineHelper} to

    95      * the provided {@code Spliterator} and send the results to the provided

    96      * {@code Sink}.

    97      *

    98      * @implSpec

    99      * The implementation behaves as if:

   100      * <pre>{@code

   101      *     intoWrapped(wrapSink(sink), spliterator);

   102      * }</pre>

   103      *

   104      * @param sink the {@code Sink} to receive the results

   105      * @param spliterator the spliterator describing the source input to process

   106      */

   107     abstract<P_IN, S extends Sink<P_OUT>> S wrapAndCopyInto(S sink, Spliterator<P_IN> spliterator);

   108 

   109     /**

   110      * Pushes elements obtained from the {@code Spliterator} into the provided

   111      * {@code Sink}.  If the stream pipeline is known to have short-circuiting

   112      * stages in it (see {@link StreamOpFlag#SHORT_CIRCUIT}), the

   113      * {@link Sink#cancellationRequested()} is checked after each

   114      * element, stopping if cancellation is requested.

   115      *

   116      * @implSpec

   117      * This method conforms to the {@code Sink} protocol of calling

   118      * {@code Sink.begin} before pushing elements, via {@code Sink.accept}, and

   119      * calling {@code Sink.end} after all elements have been pushed.

   120      *

   121      * @param wrappedSink the destination {@code Sink}

   122      * @param spliterator the source {@code Spliterator}

   123      */

   124     abstract<P_IN> void copyInto(Sink<P_IN> wrappedSink, Spliterator<P_IN> spliterator);

   125 

   126     /**

   127      * Pushes elements obtained from the {@code Spliterator} into the provided

   128      * {@code Sink}, checking {@link Sink#cancellationRequested()} after each

   129      * element, and stopping if cancellation is requested.

   130      *

   131      * @implSpec

   132      * This method conforms to the {@code Sink} protocol of calling

   133      * {@code Sink.begin} before pushing elements, via {@code Sink.accept}, and

   134      * calling {@code Sink.end} after all elements have been pushed or if

   135      * cancellation is requested.

   136      *

   137      * @param wrappedSink the destination {@code Sink}

   138      * @param spliterator the source {@code Spliterator}

   139      */

   140     abstract <P_IN> void copyIntoWithCancel(Sink<P_IN> wrappedSink, Spliterator<P_IN> spliterator);

   141 

   142     /**

   143      * Takes a {@code Sink} that accepts elements of the output type of the

   144      * {@code PipelineHelper}, and wrap it with a {@code Sink} that accepts

   145      * elements of the input type and implements all the intermediate operations

   146      * described by this {@code PipelineHelper}, delivering the result into the

   147      * provided {@code Sink}.

   148      *

   149      * @param sink the {@code Sink} to receive the results

   150      * @return a {@code Sink} that implements the pipeline stages and sends

   151      *         results to the provided {@code Sink}

   152      */

   153     abstract<P_IN> Sink<P_IN> wrapSink(Sink<P_OUT> sink);

   154 

   155     /**

   156      *

   157      * @param spliterator

   158      * @param <P_IN>

   159      * @return

   160      */

   161     abstract<P_IN> Spliterator<P_OUT> wrapSpliterator(Spliterator<P_IN> spliterator);

   162 

   163     /**

   164      * Constructs a @{link Node.Builder} compatible with the output shape of

   165      * this {@code PipelineHelper}.

   166      *

   167      * @param exactSizeIfKnown if >=0 then a builder will be created that has a

   168      *        fixed capacity of exactly sizeIfKnown elements; if < 0 then the

   169      *        builder has variable capacity.  A fixed capacity builder will fail

   170      *        if an element is added after the builder has reached capacity.

   171      * @param generator a factory function for array instances

   172      * @return a {@code Node.Builder} compatible with the output shape of this

   173      *         {@code PipelineHelper}

   174      */

   175     abstract Node.Builder<P_OUT> makeNodeBuilder(long exactSizeIfKnown,

   176                                                  IntFunction<P_OUT[]> generator);

   177 

   178     /**

   179      * Collects all output elements resulting from applying the pipeline stages

   180      * to the source {@code Spliterator} into a {@code Node}.

   181      *

   182      * @implNote

   183      * If the pipeline has no intermediate operations and the source is backed

   184      * by a {@code Node} then that {@code Node} will be returned (or flattened

   185      * and then returned). This reduces copying for a pipeline consisting of a

   186      * stateful operation followed by a terminal operation that returns an

   187      * array, such as:

   188      * <pre>{@code

   189      *     stream.sorted().toArray();

   190      * }</pre>

   191      *

   192      * @param spliterator the source {@code Spliterator}

   193      * @param flatten if true and the pipeline is a parallel pipeline then the

   194      *        {@code Node} returned will contain no children, otherwise the

   195      *        {@code Node} may represent the root in a tree that reflects the

   196      *        shape of the computation tree.

   197      * @param generator a factory function for array instances

   198      * @return the {@code Node} containing all output elements

   199      */

   200     abstract<P_IN> Node<P_OUT> evaluate(Spliterator<P_IN> spliterator,

   201                                         boolean flatten,

   202                                         IntFunction<P_OUT[]> generator);

   203 }