1 /*

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

     8  *

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

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

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

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

    13  * accompanied this code).

    14  *

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

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

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

    18  *

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

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

    21  * questions.

    22  */

    23 package java.util.stream;

    24 

    25 import java.util.Collections;

    26 import java.util.EnumSet;

    27 import java.util.PrimitiveIterator;

    28 import java.util.Set;

    29 import java.util.Spliterator;

    30 import java.util.function.Consumer;

    31 import java.util.function.DoubleConsumer;

    32 import java.util.function.Function;

    33 

    34 /**

    35  * Test scenarios for double streams.

    36  *

    37  * Each scenario is provided with a data source, a function that maps a fresh

    38  * stream (as provided by the data source) to a new stream, and a sink to

    39  * receive results.  Each scenario describes a different way of computing the

    40  * stream contents.  The test driver will ensure that all scenarios produce

    41  * the same output (modulo allowable differences in ordering).

    42  */

    43 @SuppressWarnings({"rawtypes", "unchecked"})

    44 public enum DoubleStreamTestScenario implements OpTestCase.BaseStreamTestScenario {

    45 

    46     STREAM_FOR_EACH(false) {

    47         <T, S_IN extends BaseStream<T, S_IN>>

    48         void run(TestData<T, S_IN> data, S_IN source, DoubleConsumer b, Function<S_IN, DoubleStream> m) {

    49             DoubleStream s = m.apply(source);

    50             if (s.isParallel()) {

    51                 s = s.sequential();

    52             }

    53             s.forEach(b);

    54         }

    55     },

    56 

    57     STREAM_TO_ARRAY(false) {

    58         <T, S_IN extends BaseStream<T, S_IN>>

    59         void run(TestData<T, S_IN> data, S_IN source, DoubleConsumer b, Function<S_IN, DoubleStream> m) {

    60             for (double t : m.apply(source).toArray()) {

    61                 b.accept(t);

    62             }

    63         }

    64     },

    65 

    66     STREAM_ITERATOR(false) {

    67         <T, S_IN extends BaseStream<T, S_IN>>

    68         void run(TestData<T, S_IN> data, S_IN source, DoubleConsumer b, Function<S_IN, DoubleStream> m) {

    69             for (PrimitiveIterator.OfDouble seqIter = m.apply(source).iterator(); seqIter.hasNext(); )

    70                 b.accept(seqIter.nextDouble());

    71         }

    72     },

    73 

    74     // Wrap as stream, and spliterate then iterate in pull mode

    75     STREAM_SPLITERATOR(false) {

    76         <T, S_IN extends BaseStream<T, S_IN>>

    77         void run(TestData<T, S_IN> data, S_IN source, DoubleConsumer b, Function<S_IN, DoubleStream> m) {

    78             for (Spliterator.OfDouble spl = m.apply(source).spliterator(); spl.tryAdvance(b); ) {

    79             }

    80         }

    81     },

    82 

    83     // Wrap as stream, spliterate, then split a few times mixing advances with forEach

    84     STREAM_SPLITERATOR_WITH_MIXED_TRAVERSE_AND_SPLIT(false) {

    85         <T, S_IN extends BaseStream<T, S_IN>>

    86         void run(TestData<T, S_IN> data, S_IN source, DoubleConsumer b, Function<S_IN, DoubleStream> m) {

    87             SpliteratorTestHelper.mixedTraverseAndSplit(b, m.apply(source).spliterator());

    88         }

    89     },

    90 

    91     // Wrap as stream, and spliterate then iterate in pull mode

    92     STREAM_SPLITERATOR_FOREACH(false) {

    93         <T, S_IN extends BaseStream<T, S_IN>>

    94         void run(TestData<T, S_IN> data, S_IN source, DoubleConsumer b, Function<S_IN, DoubleStream> m) {

    95             m.apply(source).spliterator().forEachRemaining(b);

    96         }

    97     },

    98 

    99     PAR_STREAM_SEQUENTIAL_FOR_EACH(true) {

   100         <T, S_IN extends BaseStream<T, S_IN>>

   101         void run(TestData<T, S_IN> data, S_IN source, DoubleConsumer b, Function<S_IN, DoubleStream> m) {

   102             m.apply(source).sequential().forEach(b);

   103         }

   104     },

   105 

   106     // Wrap as parallel stream + forEachOrdered

   107     PAR_STREAM_FOR_EACH_ORDERED(true) {

   108         <T, S_IN extends BaseStream<T, S_IN>>

   109         void run(TestData<T, S_IN> data, S_IN source, DoubleConsumer b, Function<S_IN, DoubleStream> m) {

   110             // @@@ Want to explicitly select ordered equalator

   111             m.apply(source).forEachOrdered(b);

   112         }

   113     },

   114 

   115     // Wrap as stream, and spliterate then iterate sequentially

   116     PAR_STREAM_SPLITERATOR(true) {

   117         <T, S_IN extends BaseStream<T, S_IN>>

   118         void run(TestData<T, S_IN> data, S_IN source, DoubleConsumer b, Function<S_IN, DoubleStream> m) {

   119             for (Spliterator.OfDouble spl = m.apply(source).spliterator(); spl.tryAdvance(b); ) {

   120             }

   121         }

   122     },

   123 

   124     // Wrap as stream, and spliterate then iterate sequentially

   125     PAR_STREAM_SPLITERATOR_FOREACH(true) {

   126         <T, S_IN extends BaseStream<T, S_IN>>

   127         void run(TestData<T, S_IN> data, S_IN source, DoubleConsumer b, Function<S_IN, DoubleStream> m) {

   128             m.apply(source).spliterator().forEachRemaining(b);

   129         }

   130     },

   131 

   132     PAR_STREAM_TO_ARRAY(true) {

   133         <T, S_IN extends BaseStream<T, S_IN>>

   134         void run(TestData<T, S_IN> data, S_IN source, DoubleConsumer b, Function<S_IN, DoubleStream> m) {

   135             for (double t : m.apply(source).toArray())

   136                 b.accept(t);

   137         }

   138     },

   139 

   140     // Wrap as parallel stream, get the spliterator, wrap as a stream + toArray

   141     PAR_STREAM_SPLITERATOR_STREAM_TO_ARRAY(true) {

   142         <T, S_IN extends BaseStream<T, S_IN>>

   143         void run(TestData<T, S_IN> data, S_IN source, DoubleConsumer b, Function<S_IN, DoubleStream> m) {

   144             DoubleStream s = m.apply(source);

   145             Spliterator.OfDouble sp = s.spliterator();

   146             DoubleStream ss = StreamSupport.doubleStream(() -> sp,

   147                                                          StreamOpFlag.toCharacteristics(OpTestCase.getStreamFlags(s))

   148                                                          | (sp.getExactSizeIfKnown() < 0 ? 0 : Spliterator.SIZED), true);

   149             for (double t : ss.toArray())

   150                 b.accept(t);

   151         }

   152     },

   153 

   154     PAR_STREAM_TO_ARRAY_CLEAR_SIZED(true) {

   155         <T, S_IN extends BaseStream<T, S_IN>>

   156         void run(TestData<T, S_IN> data, S_IN source, DoubleConsumer b, Function<S_IN, DoubleStream> m) {

   157             S_IN pipe1 = (S_IN) OpTestCase.chain(source,

   158                                                  new FlagDeclaringOp(StreamOpFlag.NOT_SIZED, data.getShape()));

   159             DoubleStream pipe2 = m.apply(pipe1);

   160 

   161             for (double t : pipe2.toArray())

   162                 b.accept(t);

   163         }

   164     },

   165 

   166     // Wrap as parallel stream + forEach synchronizing

   167     PAR_STREAM_FOR_EACH(true, false) {

   168         <T, S_IN extends BaseStream<T, S_IN>>

   169         void run(TestData<T, S_IN> data, S_IN source, DoubleConsumer b, Function<S_IN, DoubleStream> m) {

   170             m.apply(source).forEach(e -> {

   171                 synchronized (data) {

   172                     b.accept(e);

   173                 }

   174             });

   175         }

   176     },

   177 

   178     // Wrap as parallel stream + forEach synchronizing and clear SIZED flag

   179     PAR_STREAM_FOR_EACH_CLEAR_SIZED(true, false) {

   180         <T, S_IN extends BaseStream<T, S_IN>>

   181         void run(TestData<T, S_IN> data, S_IN source, DoubleConsumer b, Function<S_IN, DoubleStream> m) {

   182             S_IN pipe1 = (S_IN) OpTestCase.chain(source,

   183                                                  new FlagDeclaringOp(StreamOpFlag.NOT_SIZED, data.getShape()));

   184             m.apply(pipe1).forEach(e -> {

   185                 synchronized (data) {

   186                     b.accept(e);

   187                 }

   188             });

   189         }

   190     },

   191     ;

   192 

   193     // The set of scenarios that clean the SIZED flag

   194     public static final Set<DoubleStreamTestScenario> CLEAR_SIZED_SCENARIOS = Collections.unmodifiableSet(

   195             EnumSet.of(PAR_STREAM_TO_ARRAY_CLEAR_SIZED, PAR_STREAM_FOR_EACH_CLEAR_SIZED));

   196 

   197     private boolean isParallel;

   198 

   199     private final boolean isOrdered;

   200 

   201     DoubleStreamTestScenario(boolean isParallel) {

   202         this(isParallel, true);

   203     }

   204 

   205     DoubleStreamTestScenario(boolean isParallel, boolean isOrdered) {

   206         this.isParallel = isParallel;

   207         this.isOrdered = isOrdered;

   208     }

   209 

   210     public StreamShape getShape() {

   211         return StreamShape.DOUBLE_VALUE;

   212     }

   213 

   214     public boolean isParallel() {

   215         return isParallel;

   216     }

   217 

   218     public boolean isOrdered() {

   219         return isOrdered;

   220     }

   221 

   222     public <T, U, S_IN extends BaseStream<T, S_IN>, S_OUT extends BaseStream<U, S_OUT>>

   223     void run(TestData<T, S_IN> data, Consumer<U> b, Function<S_IN, S_OUT> m) {

   224         try (S_IN source = getStream(data)) {

   225             run(data, source, (DoubleConsumer) b, (Function<S_IN, DoubleStream>) m);

   226         }

   227     }

   228 

   229     abstract <T, S_IN extends BaseStream<T, S_IN>>

   230     void run(TestData<T, S_IN> data, S_IN source, DoubleConsumer b, Function<S_IN, DoubleStream> m);

   231 

   232 }