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