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.
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.Iterator;
26 import java.util.Spliterator;
27 import java.util.function.Consumer;
28 import java.util.function.Function;
29
30 /**
31 * Test scenarios for reference streams.
32 *
33 * Each scenario is provided with a data source, a function that maps a fresh
34 * stream (as provided by the data source) to a new stream, and a sink to
35 * receive results. Each scenario describes a different way of computing the
36 * stream contents. The test driver will ensure that all scenarios produce
37 * the same output (modulo allowable differences in ordering).
38 */
39 @SuppressWarnings({"rawtypes", "unchecked"})
40 public enum StreamTestScenario implements OpTestCase.BaseStreamTestScenario {
41
42 STREAM_FOR_EACH_WITH_CLOSE(false) {
43 <T, U, S_IN extends BaseStream<T, S_IN>>
44 void _run(TestData<T, S_IN> data, Consumer<U> b, Function<S_IN, Stream<U>> m) {
45 Stream<U> s = m.apply(data.stream());
46 if (s.isParallel()) {
47 s = s.sequential();
48 }
49 s.forEach(b);
50 s.close();
51 }
52 },
53
54 // Collec to list
55 STREAM_COLLECT(false) {
56 <T, U, S_IN extends BaseStream<T, S_IN>>
57 void _run(TestData<T, S_IN> data, Consumer<U> b, Function<S_IN, Stream<U>> m) {
58 for (U t : m.apply(data.stream()).collect(Collectors.toList())) {
59 b.accept(t);
60 }
61 }
62 },
63
64 // To array
65 STREAM_TO_ARRAY(false) {
66 <T, U, S_IN extends BaseStream<T, S_IN>>
67 void _run(TestData<T, S_IN> data, Consumer<U> b, Function<S_IN, Stream<U>> m) {
68 for (Object t : m.apply(data.stream()).toArray()) {
69 b.accept((U) t);
70 }
71 }
72 },
73
74 // Wrap as stream, and iterate in pull mode
75 STREAM_ITERATOR(false) {
76 <T, U, S_IN extends BaseStream<T, S_IN>>
77 void _run(TestData<T, S_IN> data, Consumer<U> b, Function<S_IN, Stream<U>> m) {
78 for (Iterator<U> seqIter = m.apply(data.stream()).iterator(); seqIter.hasNext(); )
79 b.accept(seqIter.next());
80 }
81 },
82
83 // Wrap as stream, and spliterate then iterate in pull mode
84 STREAM_SPLITERATOR(false) {
85 <T, U, S_IN extends BaseStream<T, S_IN>>
86 void _run(TestData<T, S_IN> data, Consumer<U> b, Function<S_IN, Stream<U>> m) {
87 for (Spliterator<U> spl = m.apply(data.stream()).spliterator(); spl.tryAdvance(b); ) { }
88 }
89 },
90
91 // Wrap as stream, spliterate, then split a few times mixing advances with forEach
92 STREAM_SPLITERATOR_WITH_MIXED_TRAVERSE_AND_SPLIT(false) {
93 <T, U, S_IN extends BaseStream<T, S_IN>>
94 void _run(TestData<T, S_IN> data, Consumer<U> b, Function<S_IN, Stream<U>> m) {
95 SpliteratorTestHelper.mixedTraverseAndSplit(b, m.apply(data.stream()).spliterator());
96 }
97 },
98
99 // Wrap as stream, and spliterate then iterate in pull mode
100 STREAM_SPLITERATOR_FOREACH(false) {
101 <T, U, S_IN extends BaseStream<T, S_IN>>
102 void _run(TestData<T, S_IN> data, Consumer<U> b, Function<S_IN, Stream<U>> m) {
103 m.apply(data.stream()).spliterator().forEachRemaining(b);
104 }
105 },
106
107 // Wrap as parallel stream + sequential
108 PAR_STREAM_SEQUENTIAL_FOR_EACH(true) {
109 <T, U, S_IN extends BaseStream<T, S_IN>>
110 void _run(TestData<T, S_IN> data, Consumer<U> b, Function<S_IN, Stream<U>> m) {
111 m.apply(data.parallelStream()).sequential().forEach(b);
112 }
113 },
114
115 // Wrap as parallel stream + forEachOrdered
116 PAR_STREAM_FOR_EACH_ORDERED(true) {
117 <T, U, S_IN extends BaseStream<T, S_IN>>
118 void _run(TestData<T, S_IN> data, Consumer<U> b, Function<S_IN, Stream<U>> m) {
119 // @@@ Want to explicitly select ordered equalator
120 m.apply(data.parallelStream()).forEachOrdered(b);
121 }
122 },
123
124 // Wrap as stream, and spliterate then iterate sequentially
125 PAR_STREAM_SPLITERATOR(true) {
126 <T, U, S_IN extends BaseStream<T, S_IN>>
127 void _run(TestData<T, S_IN> data, Consumer<U> b, Function<S_IN, Stream<U>> m) {
128 for (Spliterator<U> spl = m.apply(data.parallelStream()).spliterator(); spl.tryAdvance(b); ) { }
129 }
130 },
131
132 // Wrap as stream, and spliterate then iterate sequentially
133 PAR_STREAM_SPLITERATOR_FOREACH(true) {
134 <T, U, S_IN extends BaseStream<T, S_IN>>
135 void _run(TestData<T, S_IN> data, Consumer<U> b, Function<S_IN, Stream<U>> m) {
136 m.apply(data.parallelStream()).spliterator().forEachRemaining(b);
137 }
138 },
139
140 // Wrap as parallel stream + toArray
141 PAR_STREAM_TO_ARRAY(true) {
142 <T, U, S_IN extends BaseStream<T, S_IN>>
143 void _run(TestData<T, S_IN> data, Consumer<U> b, Function<S_IN, Stream<U>> m) {
144 for (Object t : m.apply(data.parallelStream()).toArray())
145 b.accept((U) t);
146 }
147 },
148
149 // Wrap as parallel stream, get the spliterator, wrap as a stream + toArray
150 PAR_STREAM_SPLITERATOR_STREAM_TO_ARRAY(true) {
151 <T, U, S_IN extends BaseStream<T, S_IN>>
152 void _run(TestData<T, S_IN> data, Consumer<U> b, Function<S_IN, Stream<U>> m) {
153 Stream<U> s = m.apply(data.parallelStream());
154 Spliterator<U> sp = s.spliterator();
155 Stream<U> ss = StreamSupport.stream(() -> sp,
156 StreamOpFlag.toCharacteristics(OpTestCase.getStreamFlags(s))
157 | (sp.getExactSizeIfKnown() < 0 ? 0 : Spliterator.SIZED), true);
158 for (Object t : ss.toArray())
159 b.accept((U) t);
160 }
161 },
162
163 // Wrap as parallel stream + toArray and clear SIZED flag
164 PAR_STREAM_TO_ARRAY_CLEAR_SIZED(true) {
165 <T, U, S_IN extends BaseStream<T, S_IN>>
166 void _run(TestData<T, S_IN> data, Consumer<U> b, Function<S_IN, Stream<U>> m) {
167 S_IN pipe1 = (S_IN) OpTestCase.chain(data.parallelStream(),
168 new FlagDeclaringOp(StreamOpFlag.NOT_SIZED, data.getShape()));
169 Stream<U> pipe2 = m.apply(pipe1);
170
171 for (Object t : pipe2.toArray())
172 b.accept((U) t);
173 }
174 },
175
176 // Wrap as parallel + collect
177 PAR_STREAM_COLLECT(true) {
178 <T, U, S_IN extends BaseStream<T, S_IN>>
179 void _run(TestData<T, S_IN> data, Consumer<U> b, Function<S_IN, Stream<U>> m) {
180 for (U u : m.apply(data.parallelStream()).collect(Collectors.toList()))
181 b.accept(u);
182 }
183 },
184
185 // Wrap sequential as parallel, + collect
186 STREAM_TO_PAR_STREAM_COLLECT(true) {
187 <T, U, S_IN extends BaseStream<T, S_IN>>
188 void _run(TestData<T, S_IN> data, Consumer<U> b, Function<S_IN, Stream<U>> m) {
189 for (U u : m.apply(data.stream().parallel()).collect(Collectors.toList()))
190 b.accept(u);
191 }
192 },
193
194 // Wrap parallel as sequential,, + collect
195 PAR_STREAM_TO_STREAM_COLLECT(true) {
196 <T, U, S_IN extends BaseStream<T, S_IN>>
197 void _run(TestData<T, S_IN> data, Consumer<U> b, Function<S_IN, Stream<U>> m) {
198 for (U u : m.apply(data.parallelStream().sequential()).collect(Collectors.toList()))
199 b.accept(u);
200 }
201 },
202 ;
203
204 private boolean isParallel;
205
206 StreamTestScenario(boolean isParallel) {
207 this.isParallel = isParallel;
208 }
209
210 public StreamShape getShape() {
211 return StreamShape.REFERENCE;
212 }
213
214 public boolean isParallel() {
215 return isParallel;
216 }
217
218 public <T, U, S_IN extends BaseStream<T, S_IN>, S_OUT extends BaseStream<U, S_OUT>>
219 void run(TestData<T, S_IN> data, Consumer<U> b, Function<S_IN, S_OUT> m) {
220 _run(data, b, (Function<S_IN, Stream<U>>) m);
221 }
222
223 abstract <T, U, S_IN extends BaseStream<T, S_IN>>
224 void _run(TestData<T, S_IN> data, Consumer<U> b, Function<S_IN, Stream<U>> m);
225
226 }