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 */
23package java.util.stream;
24
25import java.util.Collections;
26import java.util.EnumSet;
27import java.util.PrimitiveIterator;
28import java.util.Set;
29import java.util.Spliterator;
30import java.util.function.Consumer;
31import java.util.function.Function;
32import java.util.function.LongConsumer;
33
34/**
35 * Test scenarios for long 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"})
44public enum LongStreamTestScenario 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, LongConsumer b, Function<S_IN, LongStream> m) {
49            LongStream 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, LongConsumer b, Function<S_IN, LongStream> m) {
61            for (long 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, LongConsumer b, Function<S_IN, LongStream> m) {
70            for (PrimitiveIterator.OfLong seqIter = m.apply(data.stream()).iterator(); seqIter.hasNext(); )
71                b.accept(seqIter.nextLong());
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, LongConsumer b, Function<S_IN, LongStream> m) {
79            for (Spliterator.OfLong 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, LongConsumer b, Function<S_IN, LongStream> 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, LongConsumer b, Function<S_IN, LongStream> 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, LongConsumer b, Function<S_IN, LongStream> 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, LongConsumer b, Function<S_IN, LongStream> 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, LongConsumer b, Function<S_IN, LongStream> m) {
120            for (Spliterator.OfLong 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, LongConsumer b, Function<S_IN, LongStream> 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, LongConsumer b, Function<S_IN, LongStream> m) {
136            for (long 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, LongConsumer b, Function<S_IN, LongStream> m) {
145            LongStream s = m.apply(data.parallelStream());
146            Spliterator.OfLong sp = s.spliterator();
147            LongStream ss = StreamSupport.longStream(() -> sp,
148                                                     StreamOpFlag.toCharacteristics(OpTestCase.getStreamFlags(s))
149                                                     | (sp.getExactSizeIfKnown() < 0 ? 0 : Spliterator.SIZED), true);
150            for (long 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, LongConsumer b, Function<S_IN, LongStream> m) {
158            S_IN pipe1 = (S_IN) OpTestCase.chain(data.parallelStream(),
159                                                 new FlagDeclaringOp(StreamOpFlag.NOT_SIZED, data.getShape()));
160            LongStream pipe2 = m.apply(pipe1);
161
162            for (long 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, LongConsumer b, Function<S_IN, LongStream> 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, LongConsumer b, Function<S_IN, LongStream> 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<LongStreamTestScenario> 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    LongStreamTestScenario(boolean isParallel) {
203        this(isParallel, true);
204    }
205
206    LongStreamTestScenario(boolean isParallel, boolean isOrdered) {
207        this.isParallel = isParallel;
208        this.isOrdered = isOrdered;
209    }
210
211    public StreamShape getShape() {
212        return StreamShape.LONG_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, (LongConsumer) b, (Function<S_IN, LongStream>) m);
226    }
227
228    abstract <T, S_IN extends BaseStream<T, S_IN>>
229    void _run(TestData<T, S_IN> data, LongConsumer b, Function<S_IN, LongStream> m);
230
231}
232