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.IntSummaryStatistics;
28 import java.util.Objects;
29 import java.util.OptionalDouble;
30 import java.util.OptionalInt;
31 import java.util.PrimitiveIterator;
32 import java.util.Spliterator;
33 import java.util.Spliterators;
34 import java.util.function.BiConsumer;
35 import java.util.function.BinaryOperator;
36 import java.util.function.IntBinaryOperator;
37 import java.util.function.IntConsumer;
38 import java.util.function.IntFunction;
39 import java.util.function.IntPredicate;
40 import java.util.function.IntToDoubleFunction;
41 import java.util.function.IntToLongFunction;
42 import java.util.function.IntUnaryOperator;
43 import java.util.function.ObjIntConsumer;
44 import java.util.function.Supplier;
45
46 /**
47 * Abstract base class for an intermediate pipeline stage or pipeline source
48 * stage implementing whose elements are of type {@code int}.
49 *
50 * @param <E_IN> type of elements in the upstream source
51 * @since 1.8
52 */
53 abstract class IntPipeline<E_IN>
54 extends AbstractPipeline<E_IN, Integer, IntStream>
55 implements IntStream {
56
57 /**
58 * Constructor for the head of a stream pipeline.
59 *
60 * @param source {@code Supplier<Spliterator>} describing the stream source
61 * @param sourceFlags The source flags for the stream source, described in
62 * {@link StreamOpFlag}
63 * @param parallel {@code true} if the pipeline is parallel
64 */
65 IntPipeline(Supplier<? extends Spliterator<Integer>> source,
66 int sourceFlags, boolean parallel) {
67 super(source, sourceFlags, parallel);
68 }
69
70 /**
71 * Constructor for the head of a stream pipeline.
72 *
73 * @param source {@code Spliterator} describing the stream source
74 * @param sourceFlags The source flags for the stream source, described in
75 * {@link StreamOpFlag}
76 * @param parallel {@code true} if the pipeline is parallel
77 */
78 IntPipeline(Spliterator<Integer> source,
79 int sourceFlags, boolean parallel) {
80 super(source, sourceFlags, parallel);
81 }
82
83 /**
84 * Constructor for appending an intermediate operation onto an existing
85 * pipeline.
86 *
87 * @param upstream the upstream element source
88 * @param opFlags the operation flags for the new operation
89 */
90 IntPipeline(AbstractPipeline<?, E_IN, ?> upstream, int opFlags) {
91 super(upstream, opFlags);
92 }
93
94 /**
95 * Adapt a {@code Sink<Integer> to an {@code IntConsumer}, ideally simply
96 * by casting.
97 */
98 private static IntConsumer adapt(Sink<Integer> sink) {
99 if (sink instanceof IntConsumer) {
100 return (IntConsumer) sink;
101 }
102 else {
103 if (Tripwire.ENABLED)
104 Tripwire.trip(AbstractPipeline.class,
105 "using IntStream.adapt(Sink<Integer> s)");
106 return sink::accept;
107 }
108 }
109
110 /**
111 * Adapt a {@code Spliterator<Integer>} to a {@code Spliterator.OfInt}.
112 *
113 * @implNote
114 * The implementation attempts to cast to a Spliterator.OfInt, and throws an
115 * exception if this cast is not possible.
116 */
117 private static Spliterator.OfInt adapt(Spliterator<Integer> s) {
118 if (s instanceof Spliterator.OfInt) {
119 return (Spliterator.OfInt) s;
120 }
121 else {
122 if (Tripwire.ENABLED)
123 Tripwire.trip(AbstractPipeline.class,
124 "using IntStream.adapt(Spliterator<Integer> s)");
125 throw new UnsupportedOperationException("IntStream.adapt(Spliterator<Integer> s)");
126 }
127 }
128
129
130 // Shape-specific methods
131
132 @Override
133 final StreamShape getOutputShape() {
134 return StreamShape.INT_VALUE;
135 }
136
137 @Override
138 final <P_IN> Node<Integer> evaluateToNode(PipelineHelper<Integer> helper,
139 Spliterator<P_IN> spliterator,
140 boolean flattenTree,
141 IntFunction<Integer[]> generator) {
142 return Nodes.collectInt(helper, spliterator, flattenTree);
143 }
144
145 @Override
146 final <P_IN> Spliterator<Integer> wrap(PipelineHelper<Integer> ph,
147 Supplier<Spliterator<P_IN>> supplier,
148 boolean isParallel) {
149 return new StreamSpliterators.IntWrappingSpliterator<>(ph, supplier, isParallel);
150 }
151
152 @Override
153 @SuppressWarnings("unchecked")
154 final Spliterator.OfInt lazySpliterator(Supplier<? extends Spliterator<Integer>> supplier) {
155 return new StreamSpliterators.DelegatingSpliterator.OfInt((Supplier<Spliterator.OfInt>) supplier);
156 }
157
158 @Override
159 final void forEachWithCancel(Spliterator<Integer> spliterator, Sink<Integer> sink) {
160 Spliterator.OfInt spl = adapt(spliterator);
161 IntConsumer adaptedSink = adapt(sink);
162 do { } while (!sink.cancellationRequested() && spl.tryAdvance(adaptedSink));
163 }
164
165 @Override
166 final Node.Builder<Integer> makeNodeBuilder(long exactSizeIfKnown,
167 IntFunction<Integer[]> generator) {
168 return Nodes.intBuilder(exactSizeIfKnown);
169 }
170
171
172 // IntStream
173
174 @Override
175 public final PrimitiveIterator.OfInt iterator() {
176 return Spliterators.iterator(spliterator());
177 }
178
179 @Override
180 public final Spliterator.OfInt spliterator() {
181 return adapt(super.spliterator());
182 }
183
184 // Stateless intermediate ops from IntStream
185
186 @Override
187 public final LongStream asLongStream() {
188 return new LongPipeline.StatelessOp<Integer>(this, StreamShape.INT_VALUE,
189 StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT) {
190 @Override
191 Sink<Integer> opWrapSink(int flags, Sink<Long> sink) {
192 return new Sink.ChainedInt<Long>(sink) {
193 @Override
194 public void accept(int t) {
195 downstream.accept((long) t);
196 }
197 };
198 }
199 };
200 }
201
202 @Override
203 public final DoubleStream asDoubleStream() {
204 return new DoublePipeline.StatelessOp<Integer>(this, StreamShape.INT_VALUE,
205 StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT) {
206 @Override
207 Sink<Integer> opWrapSink(int flags, Sink<Double> sink) {
208 return new Sink.ChainedInt<Double>(sink) {
209 @Override
210 public void accept(int t) {
211 downstream.accept((double) t);
212 }
213 };
214 }
215 };
216 }
217
218 @Override
219 public final Stream<Integer> boxed() {
220 return mapToObj(Integer::valueOf);
221 }
222
223 @Override
224 public final IntStream map(IntUnaryOperator mapper) {
225 Objects.requireNonNull(mapper);
226 return new StatelessOp<Integer>(this, StreamShape.INT_VALUE,
227 StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT) {
228 @Override
229 Sink<Integer> opWrapSink(int flags, Sink<Integer> sink) {
230 return new Sink.ChainedInt<Integer>(sink) {
231 @Override
232 public void accept(int t) {
233 downstream.accept(mapper.applyAsInt(t));
234 }
235 };
236 }
237 };
238 }
239
240 @Override
241 public final <U> Stream<U> mapToObj(IntFunction<? extends U> mapper) {
242 Objects.requireNonNull(mapper);
243 return new ReferencePipeline.StatelessOp<Integer, U>(this, StreamShape.INT_VALUE,
244 StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT) {
245 @Override
246 Sink<Integer> opWrapSink(int flags, Sink<U> sink) {
247 return new Sink.ChainedInt<U>(sink) {
248 @Override
249 public void accept(int t) {
250 downstream.accept(mapper.apply(t));
251 }
252 };
253 }
254 };
255 }
256
257 @Override
258 public final LongStream mapToLong(IntToLongFunction mapper) {
259 Objects.requireNonNull(mapper);
260 return new LongPipeline.StatelessOp<Integer>(this, StreamShape.INT_VALUE,
261 StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT) {
262 @Override
263 Sink<Integer> opWrapSink(int flags, Sink<Long> sink) {
264 return new Sink.ChainedInt<Long>(sink) {
265 @Override
266 public void accept(int t) {
267 downstream.accept(mapper.applyAsLong(t));
268 }
269 };
270 }
271 };
272 }
273
274 @Override
275 public final DoubleStream mapToDouble(IntToDoubleFunction mapper) {
276 Objects.requireNonNull(mapper);
277 return new DoublePipeline.StatelessOp<Integer>(this, StreamShape.INT_VALUE,
278 StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT) {
279 @Override
280 Sink<Integer> opWrapSink(int flags, Sink<Double> sink) {
281 return new Sink.ChainedInt<Double>(sink) {
282 @Override
283 public void accept(int t) {
284 downstream.accept(mapper.applyAsDouble(t));
285 }
286 };
287 }
288 };
289 }
290
291 @Override
292 public final IntStream flatMap(IntFunction<? extends IntStream> mapper) {
293 return new StatelessOp<Integer>(this, StreamShape.INT_VALUE,
294 StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT | StreamOpFlag.NOT_SIZED) {
295 @Override
296 Sink<Integer> opWrapSink(int flags, Sink<Integer> sink) {
297 return new Sink.ChainedInt<Integer>(sink) {
298 @Override
299 public void begin(long size) {
300 downstream.begin(-1);
301 }
302
303 @Override
304 public void accept(int t) {
305 // We can do better that this too; optimize for depth=0 case and just grab spliterator and forEach it
306 IntStream result = mapper.apply(t);
307 if (result != null)
308 result.sequential().forEach(i -> downstream.accept(i));
309 }
310 };
311 }
312 };
313 }
314
315 @Override
316 public IntStream unordered() {
317 if (!isOrdered())
318 return this;
319 return new StatelessOp<Integer>(this, StreamShape.INT_VALUE, StreamOpFlag.NOT_ORDERED) {
320 @Override
321 Sink<Integer> opWrapSink(int flags, Sink<Integer> sink) {
322 return sink;
323 }
324 };
325 }
326
327 @Override
328 public final IntStream filter(IntPredicate predicate) {
329 Objects.requireNonNull(predicate);
330 return new StatelessOp<Integer>(this, StreamShape.INT_VALUE,
331 StreamOpFlag.NOT_SIZED) {
332 @Override
333 Sink<Integer> opWrapSink(int flags, Sink<Integer> sink) {
334 return new Sink.ChainedInt<Integer>(sink) {
335 @Override
336 public void begin(long size) {
337 downstream.begin(-1);
338 }
339
340 @Override
341 public void accept(int t) {
342 if (predicate.test(t))
343 downstream.accept(t);
344 }
345 };
346 }
347 };
348 }
349
350 @Override
351 public final IntStream peek(IntConsumer consumer) {
352 Objects.requireNonNull(consumer);
353 return new StatelessOp<Integer>(this, StreamShape.INT_VALUE,
354 0) {
355 @Override
356 Sink<Integer> opWrapSink(int flags, Sink<Integer> sink) {
357 return new Sink.ChainedInt<Integer>(sink) {
358 @Override
359 public void accept(int t) {
360 consumer.accept(t);
361 downstream.accept(t);
362 }
363 };
364 }
365 };
366 }
367
368 // Stateful intermediate ops from IntStream
369
370 private IntStream slice(long skip, long limit) {
371 return SliceOps.makeInt(this, skip, limit);
372 }
373
374 @Override
375 public final IntStream limit(long maxSize) {
376 if (maxSize < 0)
377 throw new IllegalArgumentException(Long.toString(maxSize));
378 return slice(0, maxSize);
379 }
380
381 @Override
382 public final IntStream substream(long startingOffset) {
383 if (startingOffset < 0)
384 throw new IllegalArgumentException(Long.toString(startingOffset));
385 if (startingOffset == 0)
386 return this;
387 else
388 return slice(startingOffset, -1);
389 }
390
391 @Override
392 public final IntStream substream(long startingOffset, long endingOffset) {
393 if (startingOffset < 0 || endingOffset < startingOffset)
394 throw new IllegalArgumentException(String.format("substream(%d, %d)", startingOffset, endingOffset));
395 return slice(startingOffset, endingOffset - startingOffset);
396 }
397
398 @Override
399 public final IntStream sorted() {
400 return SortedOps.makeInt(this);
401 }
402
403 @Override
404 public final IntStream distinct() {
405 // While functional and quick to implement, this approach is not very efficient.
406 // An efficient version requires an int-specific map/set implementation.
407 return boxed().distinct().mapToInt(i -> i);
408 }
409
410 // Terminal ops from IntStream
411
412 @Override
413 public void forEach(IntConsumer action) {
414 evaluate(ForEachOps.makeInt(action, false));
415 }
416
417 @Override
418 public void forEachOrdered(IntConsumer action) {
419 evaluate(ForEachOps.makeInt(action, true));
420 }
421
422 @Override
423 public final int sum() {
424 return reduce(0, Integer::sum);
425 }
426
427 @Override
428 public final OptionalInt min() {
429 return reduce(Math::min);
430 }
431
432 @Override
433 public final OptionalInt max() {
434 return reduce(Math::max);
435 }
436
437 @Override
438 public final long count() {
439 return asLongStream().map(e -> 1L).sum();
440 }
441
442 @Override
443 public final OptionalDouble average() {
444 long[] avg = collect(() -> new long[2],
445 (ll, i) -> {
446 ll[0]++;
447 ll[1] += i;
448 },
449 (ll, rr) -> {
450 ll[0] += rr[0];
451 ll[1] += rr[1];
452 });
453 return avg[0] > 0
454 ? OptionalDouble.of((double) avg[1] / avg[0])
455 : OptionalDouble.empty();
456 }
457
458 @Override
459 public final IntSummaryStatistics summaryStatistics() {
460 return collect(IntSummaryStatistics::new, IntSummaryStatistics::accept,
461 IntSummaryStatistics::combine);
462 }
463
464 @Override
465 public final int reduce(int identity, IntBinaryOperator op) {
466 return evaluate(ReduceOps.makeInt(identity, op));
467 }
468
469 @Override
470 public final OptionalInt reduce(IntBinaryOperator op) {
471 return evaluate(ReduceOps.makeInt(op));
472 }
473
474 @Override
475 public final <R> R collect(Supplier<R> resultFactory,
476 ObjIntConsumer<R> accumulator,
477 BiConsumer<R, R> combiner) {
478 BinaryOperator<R> operator = (left, right) -> {
479 combiner.accept(left, right);
480 return left;
481 };
482 return evaluate(ReduceOps.makeInt(resultFactory, accumulator, operator));
483 }
484
485 @Override
486 public final boolean anyMatch(IntPredicate predicate) {
487 return evaluate(MatchOps.makeInt(predicate, MatchOps.MatchKind.ANY));
488 }
489
490 @Override
491 public final boolean allMatch(IntPredicate predicate) {
492 return evaluate(MatchOps.makeInt(predicate, MatchOps.MatchKind.ALL));
493 }
494
495 @Override
496 public final boolean noneMatch(IntPredicate predicate) {
497 return evaluate(MatchOps.makeInt(predicate, MatchOps.MatchKind.NONE));
498 }
499
500 @Override
501 public final OptionalInt findFirst() {
502 return evaluate(FindOps.makeInt(true));
503 }
504
505 @Override
506 public final OptionalInt findAny() {
507 return evaluate(FindOps.makeInt(false));
508 }
509
510 @Override
511 public final int[] toArray() {
512 return Nodes.flattenInt((Node.OfInt) evaluateToArrayNode(Integer[]::new))
513 .asPrimitiveArray();
514 }
515
516 //
517
518 /**
519 * Source stage of an IntStream.
520 *
521 * @param <E_IN> type of elements in the upstream source
522 * @since 1.8
523 */
524 static class Head<E_IN> extends IntPipeline<E_IN> {
525 /**
526 * Constructor for the source stage of an IntStream.
527 *
528 * @param source {@code Supplier<Spliterator>} describing the stream
529 * source
530 * @param sourceFlags the source flags for the stream source, described
531 * in {@link StreamOpFlag}
532 * @param parallel {@code true} if the pipeline is parallel
533 */
534 Head(Supplier<? extends Spliterator<Integer>> source,
535 int sourceFlags, boolean parallel) {
536 super(source, sourceFlags, parallel);
537 }
538
539 /**
540 * Constructor for the source stage of an IntStream.
541 *
542 * @param source {@code Spliterator} describing the stream source
543 * @param sourceFlags the source flags for the stream source, described
544 * in {@link StreamOpFlag}
545 * @param parallel {@code true} if the pipeline is parallel
546 */
547 Head(Spliterator<Integer> source,
548 int sourceFlags, boolean parallel) {
549 super(source, sourceFlags, parallel);
550 }
551
552 @Override
553 final boolean opIsStateful() {
554 throw new UnsupportedOperationException();
555 }
556
557 @Override
558 final Sink<E_IN> opWrapSink(int flags, Sink<Integer> sink) {
559 throw new UnsupportedOperationException();
560 }
561
562 // Optimized sequential terminal operations for the head of the pipeline
563
564 @Override
565 public void forEach(IntConsumer action) {
566 if (!isParallel()) {
567 adapt(sourceStageSpliterator()).forEachRemaining(action);
568 }
569 else {
570 super.forEach(action);
571 }
572 }
573
574 @Override
575 public void forEachOrdered(IntConsumer action) {
576 if (!isParallel()) {
577 adapt(sourceStageSpliterator()).forEachRemaining(action);
578 }
579 else {
580 super.forEachOrdered(action);
581 }
582 }
583 }
584
585 /**
586 * Base class for a stateless intermediate stage of an IntStream
587 *
588 * @param <E_IN> type of elements in the upstream source
589 * @since 1.8
590 */
591 abstract static class StatelessOp<E_IN> extends IntPipeline<E_IN> {
592 /**
593 * Construct a new IntStream by appending a stateless intermediate
594 * operation to an existing stream.
595 * @param upstream The upstream pipeline stage
596 * @param inputShape The stream shape for the upstream pipeline stage
597 * @param opFlags Operation flags for the new stage
598 */
599 StatelessOp(AbstractPipeline<?, E_IN, ?> upstream,
600 StreamShape inputShape,
601 int opFlags) {
602 super(upstream, opFlags);
603 assert upstream.getOutputShape() == inputShape;
604 }
605
606 @Override
607 final boolean opIsStateful() {
608 return false;
609 }
610 }
611
612 /**
613 * Base class for a stateful intermediate stage of an IntStream.
614 *
615 * @param <E_IN> type of elements in the upstream source
616 * @since 1.8
617 */
618 abstract static class StatefulOp<E_IN> extends IntPipeline<E_IN> {
619 /**
620 * Construct a new IntStream by appending a stateful intermediate
621 * operation to an existing stream.
622 * @param upstream The upstream pipeline stage
623 * @param inputShape The stream shape for the upstream pipeline stage
624 * @param opFlags Operation flags for the new stage
625 */
626 StatefulOp(AbstractPipeline<?, E_IN, ?> upstream,
627 StreamShape inputShape,
628 int opFlags) {
629 super(upstream, opFlags);
630 assert upstream.getOutputShape() == inputShape;
631 }
632
633 @Override
634 final boolean opIsStateful() {
635 return true;
636 }
637
638 @Override
639 abstract <P_IN> Node<Integer> opEvaluateParallel(PipelineHelper<Integer> helper,
640 Spliterator<P_IN> spliterator,
641 IntFunction<Integer[]> generator);
642 }
643 }