1 /*
2 * Copyright (c) 2013, 2017, 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.SpliteratorTestHelper;
31 import java.util.function.Consumer;
32 import java.util.function.Function;
33 import java.util.function.LongConsumer;
34
35 /**
36 * Test scenarios for long streams.
37 *
38 * Each scenario is provided with a data source, a function that maps a fresh
39 * stream (as provided by the data source) to a new stream, and a sink to
40 * receive results. Each scenario describes a different way of computing the
41 * stream contents. The test driver will ensure that all scenarios produce
42 * the same output (modulo allowable differences in ordering).
43 */
44 @SuppressWarnings({"rawtypes", "unchecked"})
45 public enum LongStreamTestScenario implements OpTestCase.BaseStreamTestScenario {
46
47 STREAM_FOR_EACH(false) {
48 <T, S_IN extends BaseStream<T, S_IN>>
49 void run(TestData<T, S_IN> data, S_IN source, LongConsumer b, Function<S_IN, LongStream> m) {
50 LongStream s = m.apply(source);
51 if (s.isParallel()) {
52 s = s.sequential();
53 }
54 s.forEach(b);
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, S_IN source, LongConsumer b, Function<S_IN, LongStream> m) {
61 for (long t : m.apply(source).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, S_IN source, LongConsumer b, Function<S_IN, LongStream> m) {
70 for (PrimitiveIterator.OfLong seqIter = m.apply(source).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, S_IN source, LongConsumer b, Function<S_IN, LongStream> m) {
79 for (Spliterator.OfLong spl = m.apply(source).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, S_IN source, LongConsumer b, Function<S_IN, LongStream> m) {
88 SpliteratorTestHelper.mixedTraverseAndSplit(b, m.apply(source).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, S_IN source, LongConsumer b, Function<S_IN, LongStream> m) {
96 m.apply(source).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, S_IN source, LongConsumer b, Function<S_IN, LongStream> m) {
103 m.apply(source).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, S_IN source, LongConsumer b, Function<S_IN, LongStream> m) {
111 // @@@ Want to explicitly select ordered equalator
112 m.apply(source).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, S_IN source, LongConsumer b, Function<S_IN, LongStream> m) {
120 for (Spliterator.OfLong spl = m.apply(source).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, S_IN source, LongConsumer b, Function<S_IN, LongStream> m) {
129 m.apply(source).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, S_IN source, LongConsumer b, Function<S_IN, LongStream> m) {
136 for (long t : m.apply(source).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, S_IN source, LongConsumer b, Function<S_IN, LongStream> m) {
145 LongStream s = m.apply(source);
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, S_IN source, LongConsumer b, Function<S_IN, LongStream> m) {
158 S_IN pipe1 = (S_IN) OpTestCase.chain(source,
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, S_IN source, LongConsumer b, Function<S_IN, LongStream> m) {
171 m.apply(source).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, S_IN source, LongConsumer b, Function<S_IN, LongStream> m) {
183 S_IN pipe1 = (S_IN) OpTestCase.chain(source,
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 try (S_IN source = getStream(data)) {
226 run(data, source, (LongConsumer) b, (Function<S_IN, LongStream>) m);
227 }
228 }
229
230 abstract <T, S_IN extends BaseStream<T, S_IN>>
231 void run(TestData<T, S_IN> data, S_IN source, LongConsumer b, Function<S_IN, LongStream> m);
232
233 }