1 /*
2 * Copyright (c) 2012, 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. 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.Objects;
28 import java.util.Optional;
29 import java.util.OptionalDouble;
30 import java.util.OptionalInt;
31 import java.util.OptionalLong;
32 import java.util.Spliterator;
33 import java.util.concurrent.CountedCompleter;
34 import java.util.function.BiConsumer;
35 import java.util.function.BiFunction;
36 import java.util.function.BinaryOperator;
37 import java.util.function.DoubleBinaryOperator;
38 import java.util.function.IntBinaryOperator;
39 import java.util.function.LongBinaryOperator;
40 import java.util.function.ObjDoubleConsumer;
41 import java.util.function.ObjIntConsumer;
42 import java.util.function.ObjLongConsumer;
43 import java.util.function.Supplier;
44
45 /**
46 * Factory for creating instances of {@code TerminalOp} that implement
47 * reductions.
48 *
49 * @since 1.8
50 */
51 final class ReduceOps {
52
53 private ReduceOps() { }
54
55 /**
56 * Constructs a {@code TerminalOp} that implements a functional reduce on
57 * reference values.
58 *
59 * @param <T> the type of the input elements
60 * @param <U> the type of the result
61 * @param seed the identity element for the reduction
62 * @param reducer the accumulating function that incorporates an additional
63 * input element into the result
64 * @param combiner the combining function that combines two intermediate
65 * results
66 * @return a {@code TerminalOp} implementing the reduction
67 */
68 public static <T, U> TerminalOp<T, U>
69 makeRef(U seed, BiFunction<U, ? super T, U> reducer, BinaryOperator<U> combiner) {
70 Objects.requireNonNull(reducer);
71 Objects.requireNonNull(combiner);
72 class ReducingSink extends Box<U> implements AccumulatingSink<T, U, ReducingSink> {
73 @Override
74 public void begin(long size) {
75 state = seed;
76 }
77
78 @Override
79 public void accept(T t) {
80 state = reducer.apply(state, t);
81 }
82
83 @Override
84 public void combine(ReducingSink other) {
85 state = combiner.apply(state, other.state);
86 }
87 }
88 return new ReduceOp<T, U, ReducingSink>(StreamShape.REFERENCE) {
89 @Override
90 public ReducingSink makeSink() {
91 return new ReducingSink();
92 }
93 };
94 }
95
96 /**
97 * Constructs a {@code TerminalOp} that implements a functional reduce on
98 * reference values producing an optional reference result.
99 *
100 * @param <T> The type of the input elements, and the type of the result
101 * @param operator The reducing function
102 * @return A {@code TerminalOp} implementing the reduction
103 */
104 public static <T> TerminalOp<T, Optional<T>>
105 makeRef(BinaryOperator<T> operator) {
106 Objects.requireNonNull(operator);
107 class ReducingSink
108 implements AccumulatingSink<T, Optional<T>, ReducingSink> {
109 private boolean empty;
110 private T state;
111
112 public void begin(long size) {
113 empty = true;
114 state = null;
115 }
116
117 @Override
118 public void accept(T t) {
119 if (empty) {
120 empty = false;
121 state = t;
122 } else {
123 state = operator.apply(state, t);
124 }
125 }
126
127 @Override
128 public Optional<T> get() {
129 return empty ? Optional.empty() : Optional.of(state);
130 }
131
132 @Override
133 public void combine(ReducingSink other) {
134 if (!other.empty)
135 accept(other.state);
136 }
137 }
138 return new ReduceOp<T, Optional<T>, ReducingSink>(StreamShape.REFERENCE) {
139 @Override
140 public ReducingSink makeSink() {
141 return new ReducingSink();
142 }
143 };
144 }
145
146 /**
147 * Constructs a {@code TerminalOp} that implements a mutable reduce on
148 * reference values.
149 *
150 * @param <T> the type of the input elements
151 * @param <I> the type of the intermediate reduction result
152 * @param collector a {@code Collector} defining the reduction
153 * @return a {@code ReduceOp} implementing the reduction
154 */
155 public static <T, I> TerminalOp<T, I>
156 makeRef(Collector<? super T, I, ?> collector) {
157 Supplier<I> supplier = Objects.requireNonNull(collector).supplier();
158 BiConsumer<I, ? super T> accumulator = collector.accumulator();
159 BinaryOperator<I> combiner = collector.combiner();
160 class ReducingSink extends Box<I>
161 implements AccumulatingSink<T, I, ReducingSink> {
162 @Override
163 public void begin(long size) {
164 state = supplier.get();
165 }
166
167 @Override
168 public void accept(T t) {
169 accumulator.accept(state, t);
170 }
171
172 @Override
173 public void combine(ReducingSink other) {
174 state = combiner.apply(state, other.state);
175 }
176 }
177 return new ReduceOp<T, I, ReducingSink>(StreamShape.REFERENCE) {
178 @Override
179 public ReducingSink makeSink() {
180 return new ReducingSink();
181 }
182
183 @Override
184 public int getOpFlags() {
185 return collector.characteristics().contains(Collector.Characteristics.UNORDERED)
186 ? StreamOpFlag.NOT_ORDERED
187 : 0;
188 }
189 };
190 }
191
192 /**
193 * Constructs a {@code TerminalOp} that implements a mutable reduce on
194 * reference values.
195 *
196 * @param <T> the type of the input elements
197 * @param <R> the type of the result
198 * @param seedFactory a factory to produce a new base accumulator
199 * @param accumulator a function to incorporate an element into an
200 * accumulator
201 * @param reducer a function to combine an accumulator into another
202 * @return a {@code TerminalOp} implementing the reduction
203 */
204 public static <T, R> TerminalOp<T, R>
205 makeRef(Supplier<R> seedFactory,
206 BiConsumer<R, ? super T> accumulator,
207 BiConsumer<R,R> reducer) {
208 Objects.requireNonNull(seedFactory);
209 Objects.requireNonNull(accumulator);
210 Objects.requireNonNull(reducer);
211 class ReducingSink extends Box<R>
212 implements AccumulatingSink<T, R, ReducingSink> {
213 @Override
214 public void begin(long size) {
215 state = seedFactory.get();
216 }
217
218 @Override
219 public void accept(T t) {
220 accumulator.accept(state, t);
221 }
222
223 @Override
224 public void combine(ReducingSink other) {
225 reducer.accept(state, other.state);
226 }
227 }
228 return new ReduceOp<T, R, ReducingSink>(StreamShape.REFERENCE) {
229 @Override
230 public ReducingSink makeSink() {
231 return new ReducingSink();
232 }
233 };
234 }
235
236 /**
237 * Constructs a {@code TerminalOp} that counts the number of stream
238 * elements. If the size of the pipeline is known then count is the size
239 * and there is no need to evaluate the pipeline. If the size of the
240 * pipeline is non known then count is produced, via reduction, using a
241 * {@link CountingSink}.
242 *
243 * @param <T> the type of the input elements
244 * @return a {@code TerminalOp} implementing the counting
245 */
246 public static <T> TerminalOp<T, Long>
247 makeRefCounting() {
248 return new ReduceOp<T, Long, CountingSink<T>>(StreamShape.REFERENCE) {
249 @Override
250 public CountingSink<T> makeSink() { return new CountingSink.OfRef<>(); }
251
252 @Override
253 public <P_IN> Long evaluateSequential(PipelineHelper<T> helper,
254 Spliterator<P_IN> spliterator) {
255 if (StreamOpFlag.SIZED.isKnown(helper.getStreamAndOpFlags()))
256 return spliterator.getExactSizeIfKnown();
257 return super.evaluateSequential(helper, spliterator);
258 }
259
260 @Override
261 public <P_IN> Long evaluateParallel(PipelineHelper<T> helper,
262 Spliterator<P_IN> spliterator) {
263 if (StreamOpFlag.SIZED.isKnown(helper.getStreamAndOpFlags()))
264 return spliterator.getExactSizeIfKnown();
265 return super.evaluateParallel(helper, spliterator);
266 }
267
268 @Override
269 public int getOpFlags() {
270 return StreamOpFlag.NOT_ORDERED;
271 }
272 };
273 }
274
275 /**
276 * Constructs a {@code TerminalOp} that implements a functional reduce on
277 * {@code int} values.
278 *
279 * @param identity the identity for the combining function
280 * @param operator the combining function
281 * @return a {@code TerminalOp} implementing the reduction
282 */
283 public static TerminalOp<Integer, Integer>
284 makeInt(int identity, IntBinaryOperator operator) {
285 Objects.requireNonNull(operator);
286 class ReducingSink
287 implements AccumulatingSink<Integer, Integer, ReducingSink>, Sink.OfInt {
288 private int state;
289
290 @Override
291 public void begin(long size) {
292 state = identity;
293 }
294
295 @Override
296 public void accept(int t) {
297 state = operator.applyAsInt(state, t);
298 }
299
300 @Override
301 public Integer get() {
302 return state;
303 }
304
305 @Override
306 public void combine(ReducingSink other) {
307 accept(other.state);
308 }
309 }
310 return new ReduceOp<Integer, Integer, ReducingSink>(StreamShape.INT_VALUE) {
311 @Override
312 public ReducingSink makeSink() {
313 return new ReducingSink();
314 }
315 };
316 }
317
318 /**
319 * Constructs a {@code TerminalOp} that implements a functional reduce on
320 * {@code int} values, producing an optional integer result.
321 *
322 * @param operator the combining function
323 * @return a {@code TerminalOp} implementing the reduction
324 */
325 public static TerminalOp<Integer, OptionalInt>
326 makeInt(IntBinaryOperator operator) {
327 Objects.requireNonNull(operator);
328 class ReducingSink
329 implements AccumulatingSink<Integer, OptionalInt, ReducingSink>, Sink.OfInt {
330 private boolean empty;
331 private int state;
332
333 public void begin(long size) {
334 empty = true;
335 state = 0;
336 }
337
338 @Override
339 public void accept(int t) {
340 if (empty) {
341 empty = false;
342 state = t;
343 }
344 else {
345 state = operator.applyAsInt(state, t);
346 }
347 }
348
349 @Override
350 public OptionalInt get() {
351 return empty ? OptionalInt.empty() : OptionalInt.of(state);
352 }
353
354 @Override
355 public void combine(ReducingSink other) {
356 if (!other.empty)
357 accept(other.state);
358 }
359 }
360 return new ReduceOp<Integer, OptionalInt, ReducingSink>(StreamShape.INT_VALUE) {
361 @Override
362 public ReducingSink makeSink() {
363 return new ReducingSink();
364 }
365 };
366 }
367
368 /**
369 * Constructs a {@code TerminalOp} that implements a mutable reduce on
370 * {@code int} values.
371 *
372 * @param <R> The type of the result
373 * @param supplier a factory to produce a new accumulator of the result type
374 * @param accumulator a function to incorporate an int into an
375 * accumulator
376 * @param combiner a function to combine an accumulator into another
377 * @return A {@code ReduceOp} implementing the reduction
378 */
379 public static <R> TerminalOp<Integer, R>
380 makeInt(Supplier<R> supplier,
381 ObjIntConsumer<R> accumulator,
382 BinaryOperator<R> combiner) {
383 Objects.requireNonNull(supplier);
384 Objects.requireNonNull(accumulator);
385 Objects.requireNonNull(combiner);
386 class ReducingSink extends Box<R>
387 implements AccumulatingSink<Integer, R, ReducingSink>, Sink.OfInt {
388 @Override
389 public void begin(long size) {
390 state = supplier.get();
391 }
392
393 @Override
394 public void accept(int t) {
395 accumulator.accept(state, t);
396 }
397
398 @Override
399 public void combine(ReducingSink other) {
400 state = combiner.apply(state, other.state);
401 }
402 }
403 return new ReduceOp<Integer, R, ReducingSink>(StreamShape.INT_VALUE) {
404 @Override
405 public ReducingSink makeSink() {
406 return new ReducingSink();
407 }
408 };
409 }
410
411 /**
412 * Constructs a {@code TerminalOp} that counts the number of stream
413 * elements. If the size of the pipeline is known then count is the size
414 * and there is no need to evaluate the pipeline. If the size of the
415 * pipeline is non known then count is produced, via reduction, using a
416 * {@link CountingSink}.
417 *
418 * @return a {@code TerminalOp} implementing the counting
419 */
420 public static TerminalOp<Integer, Long>
421 makeIntCounting() {
422 return new ReduceOp<Integer, Long, CountingSink<Integer>>(StreamShape.INT_VALUE) {
423 @Override
424 public CountingSink<Integer> makeSink() { return new CountingSink.OfInt(); }
425
426 @Override
427 public <P_IN> Long evaluateSequential(PipelineHelper<Integer> helper,
428 Spliterator<P_IN> spliterator) {
429 if (StreamOpFlag.SIZED.isKnown(helper.getStreamAndOpFlags()))
430 return spliterator.getExactSizeIfKnown();
431 return super.evaluateSequential(helper, spliterator);
432 }
433
434 @Override
435 public <P_IN> Long evaluateParallel(PipelineHelper<Integer> helper,
436 Spliterator<P_IN> spliterator) {
437 if (StreamOpFlag.SIZED.isKnown(helper.getStreamAndOpFlags()))
438 return spliterator.getExactSizeIfKnown();
439 return super.evaluateParallel(helper, spliterator);
440 }
441
442 @Override
443 public int getOpFlags() {
444 return StreamOpFlag.NOT_ORDERED;
445 }
446 };
447 }
448
449 /**
450 * Constructs a {@code TerminalOp} that implements a functional reduce on
451 * {@code long} values.
452 *
453 * @param identity the identity for the combining function
454 * @param operator the combining function
455 * @return a {@code TerminalOp} implementing the reduction
456 */
457 public static TerminalOp<Long, Long>
458 makeLong(long identity, LongBinaryOperator operator) {
459 Objects.requireNonNull(operator);
460 class ReducingSink
461 implements AccumulatingSink<Long, Long, ReducingSink>, Sink.OfLong {
462 private long state;
463
464 @Override
465 public void begin(long size) {
466 state = identity;
467 }
468
469 @Override
470 public void accept(long t) {
471 state = operator.applyAsLong(state, t);
472 }
473
474 @Override
475 public Long get() {
476 return state;
477 }
478
479 @Override
480 public void combine(ReducingSink other) {
481 accept(other.state);
482 }
483 }
484 return new ReduceOp<Long, Long, ReducingSink>(StreamShape.LONG_VALUE) {
485 @Override
486 public ReducingSink makeSink() {
487 return new ReducingSink();
488 }
489 };
490 }
491
492 /**
493 * Constructs a {@code TerminalOp} that implements a functional reduce on
494 * {@code long} values, producing an optional long result.
495 *
496 * @param operator the combining function
497 * @return a {@code TerminalOp} implementing the reduction
498 */
499 public static TerminalOp<Long, OptionalLong>
500 makeLong(LongBinaryOperator operator) {
501 Objects.requireNonNull(operator);
502 class ReducingSink
503 implements AccumulatingSink<Long, OptionalLong, ReducingSink>, Sink.OfLong {
504 private boolean empty;
505 private long state;
506
507 public void begin(long size) {
508 empty = true;
509 state = 0;
510 }
511
512 @Override
513 public void accept(long t) {
514 if (empty) {
515 empty = false;
516 state = t;
517 }
518 else {
519 state = operator.applyAsLong(state, t);
520 }
521 }
522
523 @Override
524 public OptionalLong get() {
525 return empty ? OptionalLong.empty() : OptionalLong.of(state);
526 }
527
528 @Override
529 public void combine(ReducingSink other) {
530 if (!other.empty)
531 accept(other.state);
532 }
533 }
534 return new ReduceOp<Long, OptionalLong, ReducingSink>(StreamShape.LONG_VALUE) {
535 @Override
536 public ReducingSink makeSink() {
537 return new ReducingSink();
538 }
539 };
540 }
541
542 /**
543 * Constructs a {@code TerminalOp} that implements a mutable reduce on
544 * {@code long} values.
545 *
546 * @param <R> the type of the result
547 * @param supplier a factory to produce a new accumulator of the result type
548 * @param accumulator a function to incorporate an int into an
549 * accumulator
550 * @param combiner a function to combine an accumulator into another
551 * @return a {@code TerminalOp} implementing the reduction
552 */
553 public static <R> TerminalOp<Long, R>
554 makeLong(Supplier<R> supplier,
555 ObjLongConsumer<R> accumulator,
556 BinaryOperator<R> combiner) {
557 Objects.requireNonNull(supplier);
558 Objects.requireNonNull(accumulator);
559 Objects.requireNonNull(combiner);
560 class ReducingSink extends Box<R>
561 implements AccumulatingSink<Long, R, ReducingSink>, Sink.OfLong {
562 @Override
563 public void begin(long size) {
564 state = supplier.get();
565 }
566
567 @Override
568 public void accept(long t) {
569 accumulator.accept(state, t);
570 }
571
572 @Override
573 public void combine(ReducingSink other) {
574 state = combiner.apply(state, other.state);
575 }
576 }
577 return new ReduceOp<Long, R, ReducingSink>(StreamShape.LONG_VALUE) {
578 @Override
579 public ReducingSink makeSink() {
580 return new ReducingSink();
581 }
582 };
583 }
584
585 /**
586 * Constructs a {@code TerminalOp} that counts the number of stream
587 * elements. If the size of the pipeline is known then count is the size
588 * and there is no need to evaluate the pipeline. If the size of the
589 * pipeline is non known then count is produced, via reduction, using a
590 * {@link CountingSink}.
591 *
592 * @return a {@code TerminalOp} implementing the counting
593 */
594 public static TerminalOp<Long, Long>
595 makeLongCounting() {
596 return new ReduceOp<Long, Long, CountingSink<Long>>(StreamShape.LONG_VALUE) {
597 @Override
598 public CountingSink<Long> makeSink() { return new CountingSink.OfLong(); }
599
600 @Override
601 public <P_IN> Long evaluateSequential(PipelineHelper<Long> helper,
602 Spliterator<P_IN> spliterator) {
603 if (StreamOpFlag.SIZED.isKnown(helper.getStreamAndOpFlags()))
604 return spliterator.getExactSizeIfKnown();
605 return super.evaluateSequential(helper, spliterator);
606 }
607
608 @Override
609 public <P_IN> Long evaluateParallel(PipelineHelper<Long> helper,
610 Spliterator<P_IN> spliterator) {
611 if (StreamOpFlag.SIZED.isKnown(helper.getStreamAndOpFlags()))
612 return spliterator.getExactSizeIfKnown();
613 return super.evaluateParallel(helper, spliterator);
614 }
615
616 @Override
617 public int getOpFlags() {
618 return StreamOpFlag.NOT_ORDERED;
619 }
620 };
621 }
622
623 /**
624 * Constructs a {@code TerminalOp} that implements a functional reduce on
625 * {@code double} values.
626 *
627 * @param identity the identity for the combining function
628 * @param operator the combining function
629 * @return a {@code TerminalOp} implementing the reduction
630 */
631 public static TerminalOp<Double, Double>
632 makeDouble(double identity, DoubleBinaryOperator operator) {
633 Objects.requireNonNull(operator);
634 class ReducingSink
635 implements AccumulatingSink<Double, Double, ReducingSink>, Sink.OfDouble {
636 private double state;
637
638 @Override
639 public void begin(long size) {
640 state = identity;
641 }
642
643 @Override
644 public void accept(double t) {
645 state = operator.applyAsDouble(state, t);
646 }
647
648 @Override
649 public Double get() {
650 return state;
651 }
652
653 @Override
654 public void combine(ReducingSink other) {
655 accept(other.state);
656 }
657 }
658 return new ReduceOp<Double, Double, ReducingSink>(StreamShape.DOUBLE_VALUE) {
659 @Override
660 public ReducingSink makeSink() {
661 return new ReducingSink();
662 }
663 };
664 }
665
666 /**
667 * Constructs a {@code TerminalOp} that implements a functional reduce on
668 * {@code double} values, producing an optional double result.
669 *
670 * @param operator the combining function
671 * @return a {@code TerminalOp} implementing the reduction
672 */
673 public static TerminalOp<Double, OptionalDouble>
674 makeDouble(DoubleBinaryOperator operator) {
675 Objects.requireNonNull(operator);
676 class ReducingSink
677 implements AccumulatingSink<Double, OptionalDouble, ReducingSink>, Sink.OfDouble {
678 private boolean empty;
679 private double state;
680
681 public void begin(long size) {
682 empty = true;
683 state = 0;
684 }
685
686 @Override
687 public void accept(double t) {
688 if (empty) {
689 empty = false;
690 state = t;
691 }
692 else {
693 state = operator.applyAsDouble(state, t);
694 }
695 }
696
697 @Override
698 public OptionalDouble get() {
699 return empty ? OptionalDouble.empty() : OptionalDouble.of(state);
700 }
701
702 @Override
703 public void combine(ReducingSink other) {
704 if (!other.empty)
705 accept(other.state);
706 }
707 }
708 return new ReduceOp<Double, OptionalDouble, ReducingSink>(StreamShape.DOUBLE_VALUE) {
709 @Override
710 public ReducingSink makeSink() {
711 return new ReducingSink();
712 }
713 };
714 }
715
716 /**
717 * Constructs a {@code TerminalOp} that implements a mutable reduce on
718 * {@code double} values.
719 *
720 * @param <R> the type of the result
721 * @param supplier a factory to produce a new accumulator of the result type
722 * @param accumulator a function to incorporate an int into an
723 * accumulator
724 * @param combiner a function to combine an accumulator into another
725 * @return a {@code TerminalOp} implementing the reduction
726 */
727 public static <R> TerminalOp<Double, R>
728 makeDouble(Supplier<R> supplier,
729 ObjDoubleConsumer<R> accumulator,
730 BinaryOperator<R> combiner) {
731 Objects.requireNonNull(supplier);
732 Objects.requireNonNull(accumulator);
733 Objects.requireNonNull(combiner);
734 class ReducingSink extends Box<R>
735 implements AccumulatingSink<Double, R, ReducingSink>, Sink.OfDouble {
736 @Override
737 public void begin(long size) {
738 state = supplier.get();
739 }
740
741 @Override
742 public void accept(double t) {
743 accumulator.accept(state, t);
744 }
745
746 @Override
747 public void combine(ReducingSink other) {
748 state = combiner.apply(state, other.state);
749 }
750 }
751 return new ReduceOp<Double, R, ReducingSink>(StreamShape.DOUBLE_VALUE) {
752 @Override
753 public ReducingSink makeSink() {
754 return new ReducingSink();
755 }
756 };
757 }
758
759 /**
760 * Constructs a {@code TerminalOp} that counts the number of stream
761 * elements. If the size of the pipeline is known then count is the size
762 * and there is no need to evaluate the pipeline. If the size of the
763 * pipeline is non known then count is produced, via reduction, using a
764 * {@link CountingSink}.
765 *
766 * @return a {@code TerminalOp} implementing the counting
767 */
768 public static TerminalOp<Double, Long>
769 makeDoubleCounting() {
770 return new ReduceOp<Double, Long, CountingSink<Double>>(StreamShape.DOUBLE_VALUE) {
771 @Override
772 public CountingSink<Double> makeSink() { return new CountingSink.OfDouble(); }
773
774 @Override
775 public <P_IN> Long evaluateSequential(PipelineHelper<Double> helper,
776 Spliterator<P_IN> spliterator) {
777 if (StreamOpFlag.SIZED.isKnown(helper.getStreamAndOpFlags()))
778 return spliterator.getExactSizeIfKnown();
779 return super.evaluateSequential(helper, spliterator);
780 }
781
782 @Override
783 public <P_IN> Long evaluateParallel(PipelineHelper<Double> helper,
784 Spliterator<P_IN> spliterator) {
785 if (StreamOpFlag.SIZED.isKnown(helper.getStreamAndOpFlags()))
786 return spliterator.getExactSizeIfKnown();
787 return super.evaluateParallel(helper, spliterator);
788 }
789
790 @Override
791 public int getOpFlags() {
792 return StreamOpFlag.NOT_ORDERED;
793 }
794 };
795 }
796
797 /**
798 * A sink that counts elements
799 */
800 abstract static class CountingSink<T>
801 extends Box<Long>
802 implements AccumulatingSink<T, Long, CountingSink<T>> {
803 long count;
804
805 @Override
806 public void begin(long size) {
807 count = 0L;
808 }
809
810 @Override
811 public Long get() {
812 return count;
813 }
814
815 @Override
816 public void combine(CountingSink<T> other) {
817 count += other.count;
818 }
819
820 static final class OfRef<T> extends CountingSink<T> {
821 @Override
822 public void accept(T t) {
823 count++;
824 }
825 }
826
827 static final class OfInt extends CountingSink<Integer> implements Sink.OfInt {
828 @Override
829 public void accept(int t) {
830 count++;
831 }
832 }
833
834 static final class OfLong extends CountingSink<Long> implements Sink.OfLong {
835 @Override
836 public void accept(long t) {
837 count++;
838 }
839 }
840
841 static final class OfDouble extends CountingSink<Double> implements Sink.OfDouble {
842 @Override
843 public void accept(double t) {
844 count++;
845 }
846 }
847 }
848
849 /**
850 * A type of {@code TerminalSink} that implements an associative reducing
851 * operation on elements of type {@code T} and producing a result of type
852 * {@code R}.
853 *
854 * @param <T> the type of input element to the combining operation
855 * @param <R> the result type
856 * @param <K> the type of the {@code AccumulatingSink}.
857 */
858 private interface AccumulatingSink<T, R, K extends AccumulatingSink<T, R, K>>
859 extends TerminalSink<T, R> {
860 void combine(K other);
861 }
862
863 /**
864 * State box for a single state element, used as a base class for
865 * {@code AccumulatingSink} instances
866 *
867 * @param <U> The type of the state element
868 */
869 private abstract static class Box<U> {
870 U state;
871
872 Box() {} // Avoid creation of special accessor
873
874 public U get() {
875 return state;
876 }
877 }
878
879 /**
880 * A {@code TerminalOp} that evaluates a stream pipeline and sends the
881 * output into an {@code AccumulatingSink}, which performs a reduce
882 * operation. The {@code AccumulatingSink} must represent an associative
883 * reducing operation.
884 *
885 * @param <T> the output type of the stream pipeline
886 * @param <R> the result type of the reducing operation
887 * @param <S> the type of the {@code AccumulatingSink}
888 */
889 private abstract static class ReduceOp<T, R, S extends AccumulatingSink<T, R, S>>
890 implements TerminalOp<T, R> {
891 private final StreamShape inputShape;
892
893 /**
894 * Create a {@code ReduceOp} of the specified stream shape which uses
895 * the specified {@code Supplier} to create accumulating sinks.
896 *
897 * @param shape The shape of the stream pipeline
898 */
899 ReduceOp(StreamShape shape) {
900 inputShape = shape;
901 }
902
903 public abstract S makeSink();
904
905 @Override
906 public StreamShape inputShape() {
907 return inputShape;
908 }
909
910 @Override
911 public <P_IN> R evaluateSequential(PipelineHelper<T> helper,
912 Spliterator<P_IN> spliterator) {
913 return helper.wrapAndCopyInto(makeSink(), spliterator).get();
914 }
915
916 @Override
917 public <P_IN> R evaluateParallel(PipelineHelper<T> helper,
918 Spliterator<P_IN> spliterator) {
919 return new ReduceTask<>(this, helper, spliterator).invoke().get();
920 }
921 }
922
923 /**
924 * A {@code ForkJoinTask} for performing a parallel reduce operation.
925 */
926 @SuppressWarnings("serial")
927 private static final class ReduceTask<P_IN, P_OUT, R,
928 S extends AccumulatingSink<P_OUT, R, S>>
929 extends AbstractTask<P_IN, P_OUT, S, ReduceTask<P_IN, P_OUT, R, S>> {
930 private final ReduceOp<P_OUT, R, S> op;
931
932 ReduceTask(ReduceOp<P_OUT, R, S> op,
933 PipelineHelper<P_OUT> helper,
934 Spliterator<P_IN> spliterator) {
935 super(helper, spliterator);
936 this.op = op;
937 }
938
939 ReduceTask(ReduceTask<P_IN, P_OUT, R, S> parent,
940 Spliterator<P_IN> spliterator) {
941 super(parent, spliterator);
942 this.op = parent.op;
943 }
944
945 @Override
946 protected ReduceTask<P_IN, P_OUT, R, S> makeChild(Spliterator<P_IN> spliterator) {
947 return new ReduceTask<>(this, spliterator);
948 }
949
950 @Override
951 protected S doLeaf() {
952 return helper.wrapAndCopyInto(op.makeSink(), spliterator);
953 }
954
955 @Override
956 public void onCompletion(CountedCompleter<?> caller) {
957 if (!isLeaf()) {
958 S leftResult = leftChild.getLocalResult();
959 leftResult.combine(rightChild.getLocalResult());
960 setLocalResult(leftResult);
961 }
962 // GC spliterator, left and right child
963 super.onCompletion(caller);
964 }
965 }
966 }