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