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. Oracle designates this
8 * particular file as subject to the "Classpath" exception as provided
9 * by Oracle in the LICENSE file that accompanied this code.
10 *
11 * This code is distributed in the hope that it will be useful, but WITHOUT
12 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
13 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
14 * version 2 for more details (a copy is included in the LICENSE file that
15 * accompanied this code).
16 *
17 * You should have received a copy of the GNU General Public License version
18 * 2 along with this work; if not, write to the Free Software Foundation,
19 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
20 *
21 * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
22 * or visit www.oracle.com if you need additional information or have any
23 * questions.
24 */
25 package java.util.stream;
26
27 import java.util.Comparator;
28 import java.util.Iterator;
29 import java.util.Objects;
30 import java.util.Optional;
31 import java.util.Spliterator;
32 import java.util.Spliterators;
33 import java.util.function.BiConsumer;
34 import java.util.function.BiFunction;
35 import java.util.function.BinaryOperator;
36 import java.util.function.Consumer;
37 import java.util.function.DoubleConsumer;
38 import java.util.function.Function;
39 import java.util.function.IntConsumer;
40 import java.util.function.IntFunction;
41 import java.util.function.LongConsumer;
42 import java.util.function.Predicate;
43 import java.util.function.Supplier;
44 import java.util.function.ToDoubleFunction;
45 import java.util.function.ToIntFunction;
46 import java.util.function.ToLongFunction;
47
48 /**
49 * Abstract base class for an intermediate pipeline stage or pipeline source
50 * stage implementing whose elements are of type {@code U}.
51 *
52 * @param <P_IN> type of elements in the upstream source
53 * @param <P_OUT> type of elements in produced by this stage
54 *
55 * @since 1.8
56 */
57 abstract class ReferencePipeline<P_IN, P_OUT>
58 extends AbstractPipeline<P_IN, P_OUT, Stream<P_OUT>>
59 implements Stream<P_OUT> {
60
61 /**
62 * Constructor for the head of a stream pipeline.
63 *
64 * @param source {@code Supplier<Spliterator>} describing the stream source
65 * @param sourceFlags the source flags for the stream source, described in
66 * {@link StreamOpFlag}
67 * @param parallel {@code true} if the pipeline is parallel
68 */
69 ReferencePipeline(Supplier<? extends Spliterator<?>> source,
70 int sourceFlags, boolean parallel) {
71 super(source, sourceFlags, parallel);
72 }
73
74 /**
75 * Constructor for the head of a stream pipeline.
76 *
77 * @param source {@code Spliterator} describing the stream source
78 * @param sourceFlags The source flags for the stream source, described in
79 * {@link StreamOpFlag}
80 * @param parallel {@code true} if the pipeline is parallel
81 */
82 ReferencePipeline(Spliterator<?> source,
83 int sourceFlags, boolean parallel) {
84 super(source, sourceFlags, parallel);
85 }
86
87 /**
88 * Constructor for appending an intermediate operation onto an existing
89 * pipeline.
90 *
91 * @param upstream the upstream element source.
92 */
93 ReferencePipeline(AbstractPipeline<?, P_IN, ?> upstream, int opFlags) {
94 super(upstream, opFlags);
95 }
96
97 // Shape-specific methods
98
99 @Override
100 final StreamShape getOutputShape() {
101 return StreamShape.REFERENCE;
102 }
103
104 @Override
105 final <P_IN> Node<P_OUT> evaluateToNode(PipelineHelper<P_OUT> helper,
106 Spliterator<P_IN> spliterator,
107 boolean flattenTree,
108 IntFunction<P_OUT[]> generator) {
109 return Nodes.collect(helper, spliterator, flattenTree, generator);
110 }
111
112 @Override
113 final <P_IN> Spliterator<P_OUT> wrap(PipelineHelper<P_OUT> ph,
114 Supplier<Spliterator<P_IN>> supplier,
115 boolean isParallel) {
116 return new StreamSpliterators.WrappingSpliterator<>(ph, supplier, isParallel);
117 }
118
119 @Override
120 final Spliterator<P_OUT> lazySpliterator(Supplier<? extends Spliterator<P_OUT>> supplier) {
121 return new StreamSpliterators.DelegatingSpliterator<>(supplier);
122 }
123
124 @Override
125 final void forEachWithCancel(Spliterator<P_OUT> spliterator, Sink<P_OUT> sink) {
126 do { } while (!sink.cancellationRequested() && spliterator.tryAdvance(sink));
127 }
128
129 @Override
130 final Node.Builder<P_OUT> makeNodeBuilder(long exactSizeIfKnown, IntFunction<P_OUT[]> generator) {
131 return Nodes.builder(exactSizeIfKnown, generator);
132 }
133
134
135 // BaseStream
136
137 @Override
138 public final Iterator<P_OUT> iterator() {
139 return Spliterators.iterator(spliterator());
140 }
141
142
143 // Stream
144
145 // Stateless intermediate operations from Stream
146
147 @Override
148 public Stream<P_OUT> unordered() {
149 if (!isOrdered())
150 return this;
151 return new StatelessOp<P_OUT, P_OUT>(this, StreamShape.REFERENCE, StreamOpFlag.NOT_ORDERED) {
152 @Override
153 Sink<P_OUT> opWrapSink(int flags, Sink<P_OUT> sink) {
154 return sink;
155 }
156 };
157 }
158
159 @Override
160 public final Stream<P_OUT> filter(Predicate<? super P_OUT> predicate) {
161 Objects.requireNonNull(predicate);
162 return new StatelessOp<P_OUT, P_OUT>(this, StreamShape.REFERENCE,
163 StreamOpFlag.NOT_SIZED) {
164 @Override
165 Sink<P_OUT> opWrapSink(int flags, Sink<P_OUT> sink) {
166 return new Sink.ChainedReference<P_OUT>(sink) {
167 @Override
168 public void begin(long size) {
169 downstream.begin(-1);
170 }
171
172 @Override
173 public void accept(P_OUT u) {
174 if (predicate.test(u))
175 downstream.accept(u);
176 }
177 };
178 }
179 };
180 }
181
182 @Override
183 public final <R> Stream<R> map(Function<? super P_OUT, ? extends R> mapper) {
184 Objects.requireNonNull(mapper);
185 return new StatelessOp<P_OUT, R>(this, StreamShape.REFERENCE,
186 StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT) {
187 @Override
188 Sink<P_OUT> opWrapSink(int flags, Sink<R> sink) {
189 return new Sink.ChainedReference<P_OUT>(sink) {
190 @Override
191 public void accept(P_OUT u) {
192 downstream.accept(mapper.apply(u));
193 }
194 };
195 }
196 };
197 }
198
199 @Override
200 public final IntStream mapToInt(ToIntFunction<? super P_OUT> mapper) {
201 Objects.requireNonNull(mapper);
202 return new IntPipeline.StatelessOp<P_OUT>(this, StreamShape.REFERENCE,
203 StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT) {
204 @Override
205 Sink<P_OUT> opWrapSink(int flags, Sink<Integer> sink) {
206 return new Sink.ChainedReference<P_OUT>(sink) {
207 @Override
208 public void accept(P_OUT u) {
209 downstream.accept(mapper.applyAsInt(u));
210 }
211 };
212 }
213 };
214 }
215
216 @Override
217 public final LongStream mapToLong(ToLongFunction<? super P_OUT> mapper) {
218 Objects.requireNonNull(mapper);
219 return new LongPipeline.StatelessOp<P_OUT>(this, StreamShape.REFERENCE,
220 StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT) {
221 @Override
222 Sink<P_OUT> opWrapSink(int flags, Sink<Long> sink) {
223 return new Sink.ChainedReference<P_OUT>(sink) {
224 @Override
225 public void accept(P_OUT u) {
226 downstream.accept(mapper.applyAsLong(u));
227 }
228 };
229 }
230 };
231 }
232
233 @Override
234 public final DoubleStream mapToDouble(ToDoubleFunction<? super P_OUT> mapper) {
235 Objects.requireNonNull(mapper);
236 return new DoublePipeline.StatelessOp<P_OUT>(this, StreamShape.REFERENCE,
237 StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT) {
238 @Override
239 Sink<P_OUT> opWrapSink(int flags, Sink<Double> sink) {
240 return new Sink.ChainedReference<P_OUT>(sink) {
241 @Override
242 public void accept(P_OUT u) {
243 downstream.accept(mapper.applyAsDouble(u));
244 }
245 };
246 }
247 };
248 }
249
250 @Override
251 public final <R> Stream<R> flatMap(Function<? super P_OUT, ? extends Stream<? extends R>> mapper) {
252 Objects.requireNonNull(mapper);
253 // We can do better than this, by polling cancellationRequested when stream is infinite
254 return new StatelessOp<P_OUT, R>(this, StreamShape.REFERENCE,
255 StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT | StreamOpFlag.NOT_SIZED) {
256 @Override
257 Sink<P_OUT> opWrapSink(int flags, Sink<R> sink) {
258 return new Sink.ChainedReference<P_OUT>(sink) {
259 @Override
260 public void begin(long size) {
261 downstream.begin(-1);
262 }
263
264 @Override
265 public void accept(P_OUT u) {
266 // We can do better that this too; optimize for depth=0 case and just grab spliterator and forEach it
267 Stream<? extends R> result = mapper.apply(u);
268 if (result != null)
269 result.sequential().forEach(downstream);
270 }
271 };
272 }
273 };
274 }
275
276 @Override
277 public final IntStream flatMapToInt(Function<? super P_OUT, ? extends IntStream> mapper) {
278 Objects.requireNonNull(mapper);
279 // We can do better than this, by polling cancellationRequested when stream is infinite
280 return new IntPipeline.StatelessOp<P_OUT>(this, StreamShape.REFERENCE,
281 StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT | StreamOpFlag.NOT_SIZED) {
282 @Override
283 Sink<P_OUT> opWrapSink(int flags, Sink<Integer> sink) {
284 return new Sink.ChainedReference<P_OUT>(sink) {
285 IntConsumer downstreamAsInt = downstream::accept;
286 @Override
287 public void begin(long size) {
288 downstream.begin(-1);
289 }
290
291 @Override
292 public void accept(P_OUT u) {
293 // We can do better that this too; optimize for depth=0 case and just grab spliterator and forEach it
294 IntStream result = mapper.apply(u);
295 if (result != null)
296 result.sequential().forEach(downstreamAsInt);
297 }
298 };
299 }
300 };
301 }
302
303 @Override
304 public final DoubleStream flatMapToDouble(Function<? super P_OUT, ? extends DoubleStream> mapper) {
305 Objects.requireNonNull(mapper);
306 // We can do better than this, by polling cancellationRequested when stream is infinite
307 return new DoublePipeline.StatelessOp<P_OUT>(this, StreamShape.REFERENCE,
308 StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT | StreamOpFlag.NOT_SIZED) {
309 @Override
310 Sink<P_OUT> opWrapSink(int flags, Sink<Double> sink) {
311 return new Sink.ChainedReference<P_OUT>(sink) {
312 DoubleConsumer downstreamAsDouble = downstream::accept;
313 @Override
314 public void begin(long size) {
315 downstream.begin(-1);
316 }
317
318 @Override
319 public void accept(P_OUT u) {
320 // We can do better that this too; optimize for depth=0 case and just grab spliterator and forEach it
321 DoubleStream result = mapper.apply(u);
322 if (result != null)
323 result.sequential().forEach(downstreamAsDouble);
324 }
325 };
326 }
327 };
328 }
329
330 @Override
331 public final LongStream flatMapToLong(Function<? super P_OUT, ? extends LongStream> mapper) {
332 Objects.requireNonNull(mapper);
333 // We can do better than this, by polling cancellationRequested when stream is infinite
334 return new LongPipeline.StatelessOp<P_OUT>(this, StreamShape.REFERENCE,
335 StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT | StreamOpFlag.NOT_SIZED) {
336 @Override
337 Sink<P_OUT> opWrapSink(int flags, Sink<Long> sink) {
338 return new Sink.ChainedReference<P_OUT>(sink) {
339 LongConsumer downstreamAsLong = downstream::accept;
340 @Override
341 public void begin(long size) {
342 downstream.begin(-1);
343 }
344
345 @Override
346 public void accept(P_OUT u) {
347 // We can do better that this too; optimize for depth=0 case and just grab spliterator and forEach it
348 LongStream result = mapper.apply(u);
349 if (result != null)
350 result.sequential().forEach(downstreamAsLong);
351 }
352 };
353 }
354 };
355 }
356
357 @Override
358 public final Stream<P_OUT> peek(Consumer<? super P_OUT> tee) {
359 Objects.requireNonNull(tee);
360 return new StatelessOp<P_OUT, P_OUT>(this, StreamShape.REFERENCE,
361 0) {
362 @Override
363 Sink<P_OUT> opWrapSink(int flags, Sink<P_OUT> sink) {
364 return new Sink.ChainedReference<P_OUT>(sink) {
365 @Override
366 public void accept(P_OUT u) {
367 tee.accept(u);
368 downstream.accept(u);
369 }
370 };
371 }
372 };
373 }
374
375 // Stateful intermediate operations from Stream
376
377 @Override
378 public final Stream<P_OUT> distinct() {
379 return DistinctOps.makeRef(this);
380 }
381
382 @Override
383 public final Stream<P_OUT> sorted() {
384 return SortedOps.makeRef(this);
385 }
386
387 @Override
388 public final Stream<P_OUT> sorted(Comparator<? super P_OUT> comparator) {
389 return SortedOps.makeRef(this, comparator);
390 }
391
392 private Stream<P_OUT> slice(long skip, long limit) {
393 return SliceOps.makeRef(this, skip, limit);
394 }
395
396 @Override
397 public final Stream<P_OUT> limit(long maxSize) {
398 if (maxSize < 0)
399 throw new IllegalArgumentException(Long.toString(maxSize));
400 return slice(0, maxSize);
401 }
402
403 @Override
404 public final Stream<P_OUT> substream(long startingOffset) {
405 if (startingOffset < 0)
406 throw new IllegalArgumentException(Long.toString(startingOffset));
407 if (startingOffset == 0)
408 return this;
409 else
410 return slice(startingOffset, -1);
411 }
412
413 @Override
414 public final Stream<P_OUT> substream(long startingOffset, long endingOffset) {
415 if (startingOffset < 0 || endingOffset < startingOffset)
416 throw new IllegalArgumentException(String.format("substream(%d, %d)", startingOffset, endingOffset));
417 return slice(startingOffset, endingOffset - startingOffset);
418 }
419
420 // Terminal operations from Stream
421
422 @Override
423 public void forEach(Consumer<? super P_OUT> action) {
424 evaluate(ForEachOps.makeRef(action, false));
425 }
426
427 @Override
428 public void forEachOrdered(Consumer<? super P_OUT> action) {
429 evaluate(ForEachOps.makeRef(action, true));
430 }
431
432 @Override
433 @SuppressWarnings("unchecked")
434 public final <A> A[] toArray(IntFunction<A[]> generator) {
435 // Since A has no relation to U (not possible to declare that A is an upper bound of U)
436 // there will be no static type checking.
437 // Therefore use a raw type and assume A == U rather than propagating the separation of A and U
438 // throughout the code-base.
439 // The runtime type of U is never checked for equality with the component type of the runtime type of A[].
440 // Runtime checking will be performed when an element is stored in A[], thus if A is not a
441 // super type of U an ArrayStoreException will be thrown.
442 IntFunction rawGenerator = (IntFunction) generator;
443 return (A[]) Nodes.flatten(evaluateToArrayNode(rawGenerator), rawGenerator)
444 .asArray(rawGenerator);
445 }
446
447 @Override
448 public final Object[] toArray() {
449 return toArray(Object[]::new);
450 }
451
452 @Override
453 public final boolean anyMatch(Predicate<? super P_OUT> predicate) {
454 return evaluate(MatchOps.makeRef(predicate, MatchOps.MatchKind.ANY));
455 }
456
457 @Override
458 public final boolean allMatch(Predicate<? super P_OUT> predicate) {
459 return evaluate(MatchOps.makeRef(predicate, MatchOps.MatchKind.ALL));
460 }
461
462 @Override
463 public final boolean noneMatch(Predicate<? super P_OUT> predicate) {
464 return evaluate(MatchOps.makeRef(predicate, MatchOps.MatchKind.NONE));
465 }
466
467 @Override
468 public final Optional<P_OUT> findFirst() {
469 return evaluate(FindOps.makeRef(true));
470 }
471
472 @Override
473 public final Optional<P_OUT> findAny() {
474 return evaluate(FindOps.makeRef(false));
475 }
476
477 @Override
478 public final P_OUT reduce(final P_OUT identity, final BinaryOperator<P_OUT> accumulator) {
479 return evaluate(ReduceOps.makeRef(identity, accumulator, accumulator));
480 }
481
482 @Override
483 public final Optional<P_OUT> reduce(BinaryOperator<P_OUT> accumulator) {
484 return evaluate(ReduceOps.makeRef(accumulator));
485 }
486
487 @Override
488 public final <R> R reduce(R identity, BiFunction<R, ? super P_OUT, R> accumulator, BinaryOperator<R> combiner) {
489 return evaluate(ReduceOps.makeRef(identity, accumulator, combiner));
490 }
491
492 @Override
493 public final <R, A> R collect(Collector<? super P_OUT, A, R> collector) {
494 A container;
495 if (isParallel()
496 && (collector.characteristics().contains(Collector.Characteristics.CONCURRENT))
497 && (!isOrdered() || collector.characteristics().contains(Collector.Characteristics.UNORDERED))) {
498 container = collector.supplier().get();
499 BiConsumer<A, ? super P_OUT> accumulator = collector.accumulator();
500 forEach(u -> accumulator.accept(container, u));
501 }
502 else {
503 container = evaluate(ReduceOps.makeRef(collector));
504 }
505 return collector.characteristics().contains(Collector.Characteristics.IDENTITY_FINISH)
506 ? (R) container
507 : collector.finisher().apply(container);
508 }
509
510 @Override
511 public final <R> R collect(Supplier<R> resultFactory,
512 BiConsumer<R, ? super P_OUT> accumulator,
513 BiConsumer<R, R> combiner) {
514 return evaluate(ReduceOps.makeRef(resultFactory, accumulator, combiner));
515 }
516
517 @Override
518 public final Optional<P_OUT> max(Comparator<? super P_OUT> comparator) {
519 return reduce(BinaryOperator.maxBy(comparator));
520 }
521
522 @Override
523 public final Optional<P_OUT> min(Comparator<? super P_OUT> comparator) {
524 return reduce(BinaryOperator.minBy(comparator));
525
526 }
527
528 @Override
529 public final long count() {
530 return mapToLong(e -> 1L).sum();
531 }
532
533
534 //
535
536 /**
537 * Source stage of a ReferencePipeline.
538 *
539 * @param <E_IN> type of elements in the upstream source
540 * @param <E_OUT> type of elements in produced by this stage
541 * @since 1.8
542 */
543 static class Head<E_IN, E_OUT> extends ReferencePipeline<E_IN, E_OUT> {
544 /**
545 * Constructor for the source stage of a Stream.
546 *
547 * @param source {@code Supplier<Spliterator>} describing the stream
548 * source
549 * @param sourceFlags the source flags for the stream source, described
550 * in {@link StreamOpFlag}
551 */
552 Head(Supplier<? extends Spliterator<?>> source,
553 int sourceFlags, boolean parallel) {
554 super(source, sourceFlags, parallel);
555 }
556
557 /**
558 * Constructor for the source stage of a Stream.
559 *
560 * @param source {@code Spliterator} describing the stream source
561 * @param sourceFlags the source flags for the stream source, described
562 * in {@link StreamOpFlag}
563 */
564 Head(Spliterator<?> source,
565 int sourceFlags, boolean parallel) {
566 super(source, sourceFlags, parallel);
567 }
568
569 @Override
570 final boolean opIsStateful() {
571 throw new UnsupportedOperationException();
572 }
573
574 @Override
575 final Sink<E_IN> opWrapSink(int flags, Sink<E_OUT> sink) {
576 throw new UnsupportedOperationException();
577 }
578
579 // Optimized sequential terminal operations for the head of the pipeline
580
581 @Override
582 public void forEach(Consumer<? super E_OUT> action) {
583 if (!isParallel()) {
584 sourceStageSpliterator().forEachRemaining(action);
585 }
586 else {
587 super.forEach(action);
588 }
589 }
590
591 @Override
592 public void forEachOrdered(Consumer<? super E_OUT> action) {
593 if (!isParallel()) {
594 sourceStageSpliterator().forEachRemaining(action);
595 }
596 else {
597 super.forEachOrdered(action);
598 }
599 }
600 }
601
602 /**
603 * Base class for a stateless intermediate stage of a Stream.
604 *
605 * @param <E_IN> type of elements in the upstream source
606 * @param <E_OUT> type of elements in produced by this stage
607 * @since 1.8
608 */
609 abstract static class StatelessOp<E_IN, E_OUT>
610 extends ReferencePipeline<E_IN, E_OUT> {
611 /**
612 * Construct a new Stream by appending a stateless intermediate
613 * operation to an existing stream.
614 *
615 * @param upstream The upstream pipeline stage
616 * @param inputShape The stream shape for the upstream pipeline stage
617 * @param opFlags Operation flags for the new stage
618 */
619 StatelessOp(AbstractPipeline<?, E_IN, ?> upstream,
620 StreamShape inputShape,
621 int opFlags) {
622 super(upstream, opFlags);
623 assert upstream.getOutputShape() == inputShape;
624 }
625
626 @Override
627 final boolean opIsStateful() {
628 return false;
629 }
630 }
631
632 /**
633 * Base class for a stateful intermediate stage of a Stream.
634 *
635 * @param <E_IN> type of elements in the upstream source
636 * @param <E_OUT> type of elements in produced by this stage
637 * @since 1.8
638 */
639 abstract static class StatefulOp<E_IN, E_OUT>
640 extends ReferencePipeline<E_IN, E_OUT> {
641 /**
642 * Construct a new Stream by appending a stateful intermediate operation
643 * to an existing stream.
644 * @param upstream The upstream pipeline stage
645 * @param inputShape The stream shape for the upstream pipeline stage
646 * @param opFlags Operation flags for the new stage
647 */
648 StatefulOp(AbstractPipeline<?, E_IN, ?> upstream,
649 StreamShape inputShape,
650 int opFlags) {
651 super(upstream, opFlags);
652 assert upstream.getOutputShape() == inputShape;
653 }
654
655 @Override
656 final boolean opIsStateful() {
657 return true;
658 }
659
660 @Override
661 abstract <P_IN> Node<E_OUT> opEvaluateParallel(PipelineHelper<E_OUT> helper,
662 Spliterator<P_IN> spliterator,
663 IntFunction<E_OUT[]> generator);
664 }
665 }