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