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 // We can do better that this too; optimize for depth=0 case and just grab spliterator and forEach it
270 DoubleStream result = mapper.apply(t);
271 if (result != null)
272 result.sequential().forEach(i -> downstream.accept(i));
273 }
274 };
275 }
276 };
277 }
278
279 @Override
280 public DoubleStream unordered() {
281 if (!isOrdered())
282 return this;
283 return new StatelessOp<Double>(this, StreamShape.DOUBLE_VALUE, StreamOpFlag.NOT_ORDERED) {
284 @Override
285 Sink<Double> opWrapSink(int flags, Sink<Double> sink) {
286 return sink;
287 }
288 };
289 }
290
291 @Override
292 public final DoubleStream filter(DoublePredicate predicate) {
293 Objects.requireNonNull(predicate);
294 return new StatelessOp<Double>(this, StreamShape.DOUBLE_VALUE,
295 StreamOpFlag.NOT_SIZED) {
296 @Override
297 Sink<Double> opWrapSink(int flags, Sink<Double> sink) {
298 return new Sink.ChainedDouble<Double>(sink) {
299 @Override
300 public void begin(long size) {
301 downstream.begin(-1);
302 }
303
304 @Override
305 public void accept(double t) {
306 if (predicate.test(t))
307 downstream.accept(t);
308 }
309 };
310 }
311 };
312 }
313
314 @Override
315 public final DoubleStream peek(DoubleConsumer consumer) {
316 Objects.requireNonNull(consumer);
317 return new StatelessOp<Double>(this, StreamShape.DOUBLE_VALUE,
318 0) {
319 @Override
320 Sink<Double> opWrapSink(int flags, Sink<Double> sink) {
321 return new Sink.ChainedDouble<Double>(sink) {
322 @Override
323 public void accept(double t) {
324 consumer.accept(t);
325 downstream.accept(t);
326 }
327 };
328 }
329 };
330 }
331
332 // Stateful intermediate ops from DoubleStream
333
334 @Override
335 public final DoubleStream limit(long maxSize) {
336 if (maxSize < 0)
337 throw new IllegalArgumentException(Long.toString(maxSize));
338 return SliceOps.makeDouble(this, (long) 0, maxSize);
339 }
340
341 @Override
342 public final DoubleStream substream(long startingOffset) {
343 if (startingOffset < 0)
344 throw new IllegalArgumentException(Long.toString(startingOffset));
345 if (startingOffset == 0)
346 return this;
347 else {
348 long limit = -1;
349 return SliceOps.makeDouble(this, startingOffset, limit);
350 }
351 }
352
353 @Override
354 public final DoubleStream substream(long startingOffset, long endingOffset) {
355 if (startingOffset < 0 || endingOffset < startingOffset)
356 throw new IllegalArgumentException(String.format("substream(%d, %d)", startingOffset, endingOffset));
357 return SliceOps.makeDouble(this, startingOffset, endingOffset - startingOffset);
358 }
359
360 @Override
361 public final DoubleStream sorted() {
362 return SortedOps.makeDouble(this);
363 }
364
365 @Override
366 public final DoubleStream distinct() {
367 // While functional and quick to implement, this approach is not very efficient.
368 // An efficient version requires a double-specific map/set implementation.
369 return boxed().distinct().mapToDouble(i -> (double) i);
370 }
371
372 // Terminal ops from DoubleStream
373
374 @Override
375 public void forEach(DoubleConsumer consumer) {
376 evaluate(ForEachOps.makeDouble(consumer, false));
377 }
378
379 @Override
380 public void forEachOrdered(DoubleConsumer consumer) {
381 evaluate(ForEachOps.makeDouble(consumer, true));
382 }
383
384 @Override
385 public final double sum() {
386 // TODO: better algorithm to compensate for errors
387 return reduce(0.0, Double::sum);
388 }
389
390 @Override
391 public final OptionalDouble min() {
392 return reduce(Math::min);
393 }
394
395 @Override
396 public final OptionalDouble max() {
397 return reduce(Math::max);
398 }
399
400 @Override
401 public final OptionalDouble average() {
402 double[] avg = collect(() -> new double[2],
403 (ll, i) -> {
404 ll[0]++;
405 ll[1] += i;
406 },
407 (ll, rr) -> {
408 ll[0] += rr[0];
409 ll[1] += rr[1];
410 });
411 return avg[0] > 0
412 ? OptionalDouble.of(avg[1] / avg[0])
413 : OptionalDouble.empty();
414 }
415
416 @Override
417 public final long count() {
418 return mapToObj(e -> null).mapToInt(e -> 1).sum();
419 }
420
421 @Override
422 public final DoubleSummaryStatistics summaryStatistics() {
423 return collect(DoubleSummaryStatistics::new, DoubleSummaryStatistics::accept,
424 DoubleSummaryStatistics::combine);
425 }
426
427 @Override
428 public final double reduce(double identity, DoubleBinaryOperator op) {
429 return evaluate(ReduceOps.makeDouble(identity, op));
430 }
431
432 @Override
433 public final OptionalDouble reduce(DoubleBinaryOperator op) {
434 return evaluate(ReduceOps.makeDouble(op));
435 }
436
437 @Override
438 public final <R> R collect(Supplier<R> resultFactory,
439 ObjDoubleConsumer<R> accumulator,
440 BiConsumer<R, R> combiner) {
441 BinaryOperator<R> operator = (left, right) -> {
442 combiner.accept(left, right);
443 return left;
444 };
445 return evaluate(ReduceOps.makeDouble(resultFactory, accumulator, operator));
446 }
447
448 @Override
449 public final boolean anyMatch(DoublePredicate predicate) {
450 return evaluate(MatchOps.makeDouble(predicate, MatchOps.MatchKind.ANY));
451 }
452
453 @Override
454 public final boolean allMatch(DoublePredicate predicate) {
455 return evaluate(MatchOps.makeDouble(predicate, MatchOps.MatchKind.ALL));
456 }
457
458 @Override
459 public final boolean noneMatch(DoublePredicate predicate) {
460 return evaluate(MatchOps.makeDouble(predicate, MatchOps.MatchKind.NONE));
461 }
462
463 @Override
464 public final OptionalDouble findFirst() {
465 return evaluate(FindOps.makeDouble(true));
466 }
467
468 @Override
469 public final OptionalDouble findAny() {
470 return evaluate(FindOps.makeDouble(false));
471 }
472
473 @Override
474 public final double[] toArray() {
475 return Nodes.flattenDouble((Node.OfDouble) evaluateToArrayNode(Double[]::new))
476 .asPrimitiveArray();
477 }
478
479 //
480
481 /**
482 * Source stage of a DoubleStream
483 *
484 * @param <E_IN> type of elements in the upstream source
485 */
486 static class Head<E_IN> extends DoublePipeline<E_IN> {
487 /**
488 * Constructor for the source stage of a DoubleStream.
489 *
490 * @param source {@code Supplier<Spliterator>} describing the stream
491 * source
492 * @param sourceFlags the source flags for the stream source, described
493 * in {@link StreamOpFlag}
494 * @param parallel {@code true} if the pipeline is parallel
495 */
496 Head(Supplier<? extends Spliterator<Double>> source,
497 int sourceFlags, boolean parallel) {
498 super(source, sourceFlags, parallel);
499 }
500
501 /**
502 * Constructor for the source stage of a DoubleStream.
503 *
504 * @param source {@code Spliterator} describing the stream source
505 * @param sourceFlags the source flags for the stream source, described
506 * in {@link StreamOpFlag}
507 * @param parallel {@code true} if the pipeline is parallel
508 */
509 Head(Spliterator<Double> source,
510 int sourceFlags, boolean parallel) {
511 super(source, sourceFlags, parallel);
512 }
513
514 @Override
515 final boolean opIsStateful() {
516 throw new UnsupportedOperationException();
517 }
518
519 @Override
520 final Sink<E_IN> opWrapSink(int flags, Sink<Double> sink) {
521 throw new UnsupportedOperationException();
522 }
523
524 // Optimized sequential terminal operations for the head of the pipeline
525
526 @Override
527 public void forEach(DoubleConsumer consumer) {
528 if (!isParallel()) {
529 adapt(sourceStageSpliterator()).forEachRemaining(consumer);
530 }
531 else {
532 super.forEach(consumer);
533 }
534 }
535
536 @Override
537 public void forEachOrdered(DoubleConsumer consumer) {
538 if (!isParallel()) {
539 adapt(sourceStageSpliterator()).forEachRemaining(consumer);
540 }
541 else {
542 super.forEachOrdered(consumer);
543 }
544 }
545
546 }
547
548 /**
549 * Base class for a stateless intermediate stage of a DoubleStream.
550 *
551 * @param <E_IN> type of elements in the upstream source
552 * @since 1.8
553 */
554 abstract static class StatelessOp<E_IN> extends DoublePipeline<E_IN> {
555 /**
556 * Construct a new DoubleStream by appending a stateless intermediate
557 * operation to an existing stream.
558 *
559 * @param upstream the upstream pipeline stage
560 * @param inputShape the stream shape for the upstream pipeline stage
561 * @param opFlags operation flags for the new stage
562 */
563 StatelessOp(AbstractPipeline<?, E_IN, ?> upstream,
564 StreamShape inputShape,
565 int opFlags) {
566 super(upstream, opFlags);
567 assert upstream.getOutputShape() == inputShape;
568 }
569
570 @Override
571 final boolean opIsStateful() {
572 return false;
573 }
574 }
575
576 /**
577 * Base class for a stateful intermediate stage of a DoubleStream.
578 *
579 * @param <E_IN> type of elements in the upstream source
580 * @since 1.8
581 */
582 abstract static class StatefulOp<E_IN> extends DoublePipeline<E_IN> {
583 /**
584 * Construct a new DoubleStream by appending a stateful intermediate
585 * operation to an existing stream.
586 *
587 * @param upstream the upstream pipeline stage
588 * @param inputShape the stream shape for the upstream pipeline stage
589 * @param opFlags operation flags for the new stage
590 */
591 StatefulOp(AbstractPipeline<?, E_IN, ?> upstream,
592 StreamShape inputShape,
593 int opFlags) {
594 super(upstream, opFlags);
595 assert upstream.getOutputShape() == inputShape;
596 }
597
598 @Override
599 final boolean opIsStateful() {
600 return true;
601 }
602
603 @Override
604 abstract <P_IN> Node<Double> opEvaluateParallel(PipelineHelper<Double> helper,
605 Spliterator<P_IN> spliterator,
606 IntFunction<Double[]> generator);
607 }
608 }