Home | History | Annotate | Download | only in stream
      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_WITH_CLOSE(false) {
     47         <T, S_IN extends BaseStream<T, S_IN>>
     48         void _run(TestData<T, S_IN> data, DoubleConsumer b, Function<S_IN, DoubleStream> m) {
     49             DoubleStream s = m.apply(data.stream());
     50             if (s.isParallel()) {
     51                 s = s.sequential();
     52             }
     53             s.forEach(b);
     54             s.close();
     55         }
     56     },
     57 
     58     STREAM_TO_ARRAY(false) {
     59         <T, S_IN extends BaseStream<T, S_IN>>
     60         void _run(TestData<T, S_IN> data, DoubleConsumer b, Function<S_IN, DoubleStream> m) {
     61             for (double t : m.apply(data.stream()).toArray()) {
     62                 b.accept(t);
     63             }
     64         }
     65     },
     66 
     67     STREAM_ITERATOR(false) {
     68         <T, S_IN extends BaseStream<T, S_IN>>
     69         void _run(TestData<T, S_IN> data, DoubleConsumer b, Function<S_IN, DoubleStream> m) {
     70             for (PrimitiveIterator.OfDouble seqIter = m.apply(data.stream()).iterator(); seqIter.hasNext(); )
     71                 b.accept(seqIter.nextDouble());
     72         }
     73     },
     74 
     75     // Wrap as stream, and spliterate then iterate in pull mode
     76     STREAM_SPLITERATOR(false) {
     77         <T, S_IN extends BaseStream<T, S_IN>>
     78         void _run(TestData<T, S_IN> data, DoubleConsumer b, Function<S_IN, DoubleStream> m) {
     79             for (Spliterator.OfDouble spl = m.apply(data.stream()).spliterator(); spl.tryAdvance(b); ) {
     80             }
     81         }
     82     },
     83 
     84     // Wrap as stream, spliterate, then split a few times mixing advances with forEach
     85     STREAM_SPLITERATOR_WITH_MIXED_TRAVERSE_AND_SPLIT(false) {
     86         <T, S_IN extends BaseStream<T, S_IN>>
     87         void _run(TestData<T, S_IN> data, DoubleConsumer b, Function<S_IN, DoubleStream> m) {
     88             SpliteratorTestHelper.mixedTraverseAndSplit(b, m.apply(data.stream()).spliterator());
     89         }
     90     },
     91 
     92     // Wrap as stream, and spliterate then iterate in pull mode
     93     STREAM_SPLITERATOR_FOREACH(false) {
     94         <T, S_IN extends BaseStream<T, S_IN>>
     95         void _run(TestData<T, S_IN> data, DoubleConsumer b, Function<S_IN, DoubleStream> m) {
     96             m.apply(data.stream()).spliterator().forEachRemaining(b);
     97         }
     98     },
     99 
    100     PAR_STREAM_SEQUENTIAL_FOR_EACH(true) {
    101         <T, S_IN extends BaseStream<T, S_IN>>
    102         void _run(TestData<T, S_IN> data, DoubleConsumer b, Function<S_IN, DoubleStream> m) {
    103             m.apply(data.parallelStream()).sequential().forEach(b);
    104         }
    105     },
    106 
    107     // Wrap as parallel stream + forEachOrdered
    108     PAR_STREAM_FOR_EACH_ORDERED(true) {
    109         <T, S_IN extends BaseStream<T, S_IN>>
    110         void _run(TestData<T, S_IN> data, DoubleConsumer b, Function<S_IN, DoubleStream> m) {
    111             // @@@ Want to explicitly select ordered equalator
    112             m.apply(data.parallelStream()).forEachOrdered(b);
    113         }
    114     },
    115 
    116     // Wrap as stream, and spliterate then iterate sequentially
    117     PAR_STREAM_SPLITERATOR(true) {
    118         <T, S_IN extends BaseStream<T, S_IN>>
    119         void _run(TestData<T, S_IN> data, DoubleConsumer b, Function<S_IN, DoubleStream> m) {
    120             for (Spliterator.OfDouble spl = m.apply(data.parallelStream()).spliterator(); spl.tryAdvance(b); ) {
    121             }
    122         }
    123     },
    124 
    125     // Wrap as stream, and spliterate then iterate sequentially
    126     PAR_STREAM_SPLITERATOR_FOREACH(true) {
    127         <T, S_IN extends BaseStream<T, S_IN>>
    128         void _run(TestData<T, S_IN> data, DoubleConsumer b, Function<S_IN, DoubleStream> m) {
    129             m.apply(data.parallelStream()).spliterator().forEachRemaining(b);
    130         }
    131     },
    132 
    133     PAR_STREAM_TO_ARRAY(true) {
    134         <T, S_IN extends BaseStream<T, S_IN>>
    135         void _run(TestData<T, S_IN> data, DoubleConsumer b, Function<S_IN, DoubleStream> m) {
    136             for (double t : m.apply(data.parallelStream()).toArray())
    137                 b.accept(t);
    138         }
    139     },
    140 
    141     // Wrap as parallel stream, get the spliterator, wrap as a stream + toArray
    142     PAR_STREAM_SPLITERATOR_STREAM_TO_ARRAY(true) {
    143         <T, S_IN extends BaseStream<T, S_IN>>
    144         void _run(TestData<T, S_IN> data, DoubleConsumer b, Function<S_IN, DoubleStream> m) {
    145             DoubleStream s = m.apply(data.parallelStream());
    146             Spliterator.OfDouble sp = s.spliterator();
    147             DoubleStream ss = StreamSupport.doubleStream(() -> sp,
    148                                                          StreamOpFlag.toCharacteristics(OpTestCase.getStreamFlags(s))
    149                                                          | (sp.getExactSizeIfKnown() < 0 ? 0 : Spliterator.SIZED), true);
    150             for (double t : ss.toArray())
    151                 b.accept(t);
    152         }
    153     },
    154 
    155     PAR_STREAM_TO_ARRAY_CLEAR_SIZED(true) {
    156         <T, S_IN extends BaseStream<T, S_IN>>
    157         void _run(TestData<T, S_IN> data, DoubleConsumer b, Function<S_IN, DoubleStream> m) {
    158             S_IN pipe1 = (S_IN) OpTestCase.chain(data.parallelStream(),
    159                                                  new FlagDeclaringOp(StreamOpFlag.NOT_SIZED, data.getShape()));
    160             DoubleStream pipe2 = m.apply(pipe1);
    161 
    162             for (double t : pipe2.toArray())
    163                 b.accept(t);
    164         }
    165     },
    166 
    167     // Wrap as parallel stream + forEach synchronizing
    168     PAR_STREAM_FOR_EACH(true, false) {
    169         <T, S_IN extends BaseStream<T, S_IN>>
    170         void _run(TestData<T, S_IN> data, DoubleConsumer b, Function<S_IN, DoubleStream> m) {
    171             m.apply(data.parallelStream()).forEach(e -> {
    172                 synchronized (data) {
    173                     b.accept(e);
    174                 }
    175             });
    176         }
    177     },
    178 
    179     // Wrap as parallel stream + forEach synchronizing and clear SIZED flag
    180     PAR_STREAM_FOR_EACH_CLEAR_SIZED(true, false) {
    181         <T, S_IN extends BaseStream<T, S_IN>>
    182         void _run(TestData<T, S_IN> data, DoubleConsumer b, Function<S_IN, DoubleStream> m) {
    183             S_IN pipe1 = (S_IN) OpTestCase.chain(data.parallelStream(),
    184                                                  new FlagDeclaringOp(StreamOpFlag.NOT_SIZED, data.getShape()));
    185             m.apply(pipe1).forEach(e -> {
    186                 synchronized (data) {
    187                     b.accept(e);
    188                 }
    189             });
    190         }
    191     },
    192     ;
    193 
    194     // The set of scenarios that clean the SIZED flag
    195     public static final Set<DoubleStreamTestScenario> CLEAR_SIZED_SCENARIOS = Collections.unmodifiableSet(
    196             EnumSet.of(PAR_STREAM_TO_ARRAY_CLEAR_SIZED, PAR_STREAM_FOR_EACH_CLEAR_SIZED));
    197 
    198     private boolean isParallel;
    199 
    200     private final boolean isOrdered;
    201 
    202     DoubleStreamTestScenario(boolean isParallel) {
    203         this(isParallel, true);
    204     }
    205 
    206     DoubleStreamTestScenario(boolean isParallel, boolean isOrdered) {
    207         this.isParallel = isParallel;
    208         this.isOrdered = isOrdered;
    209     }
    210 
    211     public StreamShape getShape() {
    212         return StreamShape.DOUBLE_VALUE;
    213     }
    214 
    215     public boolean isParallel() {
    216         return isParallel;
    217     }
    218 
    219     public boolean isOrdered() {
    220         return isOrdered;
    221     }
    222 
    223     public <T, U, S_IN extends BaseStream<T, S_IN>, S_OUT extends BaseStream<U, S_OUT>>
    224     void run(TestData<T, S_IN> data, Consumer<U> b, Function<S_IN, S_OUT> m) {
    225         _run(data, (DoubleConsumer) b, (Function<S_IN, DoubleStream>) m);
    226     }
    227 
    228     abstract <T, S_IN extends BaseStream<T, S_IN>>
    229     void _run(TestData<T, S_IN> data, DoubleConsumer b, Function<S_IN, DoubleStream> m);
    230 
    231 }
    232