1 /*
2  * Copyright (c) 2012, 2017, 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.Comparator;
28 import java.util.Iterator;
29 import java.util.Objects;
30 import java.util.Optional;
31 import java.util.Spliterator;
32 import java.util.Spliterators;
33 import java.util.function.BiConsumer;
34 import java.util.function.BiFunction;
35 import java.util.function.BinaryOperator;
36 import java.util.function.Consumer;
37 import java.util.function.DoubleConsumer;
38 import java.util.function.Function;
39 import java.util.function.IntConsumer;
40 import java.util.function.IntFunction;
41 import java.util.function.LongConsumer;
42 import java.util.function.Predicate;
43 import java.util.function.Supplier;
44 import java.util.function.ToDoubleFunction;
45 import java.util.function.ToIntFunction;
46 import java.util.function.ToLongFunction;
47
48 /**
49  * Abstract base class for an intermediate pipeline stage or pipeline source
50  * stage implementing whose elements are of type {@code U}.
51  *
52  * @param <P_IN> type of elements in the upstream source
53  * @param <P_OUT> type of elements in produced by this stage
54  *
55  * @since 1.8
56  */
57 abstract class ReferencePipeline<P_IN, P_OUT>
58         extends AbstractPipeline<P_IN, P_OUT, Stream<P_OUT>>
59         implements Stream<P_OUT>  {
60
61     /**
62      * Constructor for the head of a stream pipeline.
63      *
64      * @param source {@code Supplier<Spliterator>} describing the stream source
65      * @param sourceFlags the source flags for the stream source, described in
66      *        {@link StreamOpFlag}
67      * @param parallel {@code true} if the pipeline is parallel
68      */
69     ReferencePipeline(Supplier<? extends Spliterator<?>> source,
70                       int sourceFlags, boolean parallel) {
71         super(source, sourceFlags, parallel);
72     }
73
74     /**
75      * Constructor for the head of a stream pipeline.
76      *
77      * @param source {@code Spliterator} describing the stream source
78      * @param sourceFlags The source flags for the stream source, described in
79      *        {@link StreamOpFlag}
80      * @param parallel {@code true} if the pipeline is parallel
81      */
82     ReferencePipeline(Spliterator<?> source,
83                       int sourceFlags, boolean parallel) {
84         super(source, sourceFlags, parallel);
85     }
86
87     /**
88      * Constructor for appending an intermediate operation onto an existing
89      * pipeline.
90      *
91      * @param upstream the upstream element source.
92      */
93     ReferencePipeline(AbstractPipeline<?, P_IN, ?> upstream, int opFlags) {
94         super(upstream, opFlags);
95     }
96
97     // Shape-specific methods
98
99     @Override
100     final StreamShape getOutputShape() {
101         return StreamShape.REFERENCE;
102     }
103
104     @Override
105     final <P_IN> Node<P_OUT> evaluateToNode(PipelineHelper<P_OUT> helper,
106                                         Spliterator<P_IN> spliterator,
107                                         boolean flattenTree,
108                                         IntFunction<P_OUT[]> generator) {
109         return Nodes.collect(helper, spliterator, flattenTree, generator);
110     }
111
112     @Override
113     final <P_IN> Spliterator<P_OUT> wrap(PipelineHelper<P_OUT> ph,
114                                      Supplier<Spliterator<P_IN>> supplier,
115                                      boolean isParallel) {
116         return new StreamSpliterators.WrappingSpliterator<>(ph, supplier, isParallel);
117     }
118
119     @Override
120     final Spliterator<P_OUT> lazySpliterator(Supplier<? extends Spliterator<P_OUT>> supplier) {
121         return new StreamSpliterators.DelegatingSpliterator<>(supplier);
122     }
123
124     @Override
125     final boolean forEachWithCancel(Spliterator<P_OUT> spliterator, Sink<P_OUT> sink) {
126         boolean cancelled;
127         do { } while (!(cancelled = sink.cancellationRequested()) && spliterator.tryAdvance(sink));
128         return cancelled;
129     }
130
131     @Override
132     final Node.Builder<P_OUT> makeNodeBuilder(long exactSizeIfKnown, IntFunction<P_OUT[]> generator) {
133         return Nodes.builder(exactSizeIfKnown, generator);
134     }
135
136
137     // BaseStream
138
139     @Override
140     public final Iterator<P_OUT> iterator() {
141         return Spliterators.iterator(spliterator());
142     }
143
144
145     // Stream
146
147     // Stateless intermediate operations from Stream
148
149     @Override
150     public Stream<P_OUT> unordered() {
151         if (!isOrdered())
152             return this;
153         return new StatelessOp<P_OUT, P_OUT>(this, StreamShape.REFERENCE, StreamOpFlag.NOT_ORDERED) {
154             @Override
155             Sink<P_OUT> opWrapSink(int flags, Sink<P_OUT> sink) {
156                 return sink;
157             }
158         };
159     }
160
161     @Override
162     public final Stream<P_OUT> filter(Predicate<? super P_OUT> predicate) {
163         Objects.requireNonNull(predicate);
164         return new StatelessOp<P_OUT, P_OUT>(this, StreamShape.REFERENCE,
165                                      StreamOpFlag.NOT_SIZED) {
166             @Override
167             Sink<P_OUT> opWrapSink(int flags, Sink<P_OUT> sink) {
168                 return new Sink.ChainedReference<P_OUT, P_OUT>(sink) {
169                     @Override
170                     public void begin(long size) {
171                         downstream.begin(-1);
172                     }
173
174                     @Override
175                     public void accept(P_OUT u) {
176                         if (predicate.test(u))
177                             downstream.accept(u);
178                     }
179                 };
180             }
181         };
182     }
183
184     @Override
185     @SuppressWarnings("unchecked")
186     public final <R> Stream<R> map(Function<? super P_OUT, ? extends R> mapper) {
187         Objects.requireNonNull(mapper);
188         return new StatelessOp<P_OUT, R>(this, StreamShape.REFERENCE,
189                                      StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT) {
190             @Override
191             Sink<P_OUT> opWrapSink(int flags, Sink<R> sink) {
192                 return new Sink.ChainedReference<P_OUT, R>(sink) {
193                     @Override
194                     public void accept(P_OUT u) {
195                         downstream.accept(mapper.apply(u));
196                     }
197                 };
198             }
199         };
200     }
201
202     @Override
203     public final IntStream mapToInt(ToIntFunction<? super P_OUT> mapper) {
204         Objects.requireNonNull(mapper);
205         return new IntPipeline.StatelessOp<P_OUT>(this, StreamShape.REFERENCE,
206                                               StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT) {
207             @Override
208             Sink<P_OUT> opWrapSink(int flags, Sink<Integer> sink) {
209                 return new Sink.ChainedReference<P_OUT, Integer>(sink) {
210                     @Override
211                     public void accept(P_OUT u) {
212                         downstream.accept(mapper.applyAsInt(u));
213                     }
214                 };
215             }
216         };
217     }
218
219     @Override
220     public final LongStream mapToLong(ToLongFunction<? super P_OUT> mapper) {
221         Objects.requireNonNull(mapper);
222         return new LongPipeline.StatelessOp<P_OUT>(this, StreamShape.REFERENCE,
223                                       StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT) {
224             @Override
225             Sink<P_OUT> opWrapSink(int flags, Sink<Long> sink) {
226                 return new Sink.ChainedReference<P_OUT, Long>(sink) {
227                     @Override
228                     public void accept(P_OUT u) {
229                         downstream.accept(mapper.applyAsLong(u));
230                     }
231                 };
232             }
233         };
234     }
235
236     @Override
237     public final DoubleStream mapToDouble(ToDoubleFunction<? super P_OUT> mapper) {
238         Objects.requireNonNull(mapper);
239         return new DoublePipeline.StatelessOp<P_OUT>(this, StreamShape.REFERENCE,
240                                         StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT) {
241             @Override
242             Sink<P_OUT> opWrapSink(int flags, Sink<Double> sink) {
243                 return new Sink.ChainedReference<P_OUT, Double>(sink) {
244                     @Override
245                     public void accept(P_OUT u) {
246                         downstream.accept(mapper.applyAsDouble(u));
247                     }
248                 };
249             }
250         };
251     }
252
253     @Override
254     public final <R> Stream<R> flatMap(Function<? super P_OUT, ? extends Stream<? extends R>> mapper) {
255         Objects.requireNonNull(mapper);
256         return new StatelessOp<P_OUT, R>(this, StreamShape.REFERENCE,
257                                      StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT | StreamOpFlag.NOT_SIZED) {
258             @Override
259             Sink<P_OUT> opWrapSink(int flags, Sink<R> sink) {
260                 return new Sink.ChainedReference<P_OUT, R>(sink) {
261                     // true if cancellationRequested() has been called
262                     boolean cancellationRequestedCalled;
263
264                     @Override
265                     public void begin(long size) {
266                         downstream.begin(-1);
267                     }
268
269                     @Override
270                     public void accept(P_OUT u) {
271                         try (Stream<? extends R> result = mapper.apply(u)) {
272                             if (result != null) {
273                                 if (!cancellationRequestedCalled) {
274                                     result.sequential().forEach(downstream);
275                                 }
276                                 else {
277                                     var s = result.sequential().spliterator();
278                                     do { } while (!downstream.cancellationRequested() && s.tryAdvance(downstream));
279                                 }
280                             }
281                         }
282                     }
283
284                     @Override
285                     public boolean cancellationRequested() {
286                         // If this method is called then an operation within the stream
287                         // pipeline is short-circuiting (see AbstractPipeline.copyInto).
288                         // Note that we cannot differentiate between an upstream or
289                         // downstream operation
290                         cancellationRequestedCalled = true;
291                         return downstream.cancellationRequested();
292                     }
293                 };
294             }
295         };
296     }
297
298     @Override
299     public final IntStream flatMapToInt(Function<? super P_OUT, ? extends IntStream> mapper) {
300         Objects.requireNonNull(mapper);
301         return new IntPipeline.StatelessOp<P_OUT>(this, StreamShape.REFERENCE,
302                                               StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT | StreamOpFlag.NOT_SIZED) {
303             @Override
304             Sink<P_OUT> opWrapSink(int flags, Sink<Integer> sink) {
305                 return new Sink.ChainedReference<P_OUT, Integer>(sink) {
306                     // true if cancellationRequested() has been called
307                     boolean cancellationRequestedCalled;
308
309                     // cache the consumer to avoid creation on every accepted element
310                     IntConsumer downstreamAsInt = downstream::accept;
311
312                     @Override
313                     public void begin(long size) {
314                         downstream.begin(-1);
315                     }
316
317                     @Override
318                     public void accept(P_OUT u) {
319                         try (IntStream result = mapper.apply(u)) {
320                             if (result != null) {
321                                 if (!cancellationRequestedCalled) {
322                                     result.sequential().forEach(downstreamAsInt);
323                                 }
324                                 else {
325                                     var s = result.sequential().spliterator();
326                                     do { } while (!downstream.cancellationRequested() && s.tryAdvance(downstreamAsInt));
327                                 }
328                             }
329                         }
330                     }
331
332                     @Override
333                     public boolean cancellationRequested() {
334                         cancellationRequestedCalled = true;
335                         return downstream.cancellationRequested();
336                     }
337                 };
338             }
339         };
340     }
341
342     @Override
343     public final DoubleStream flatMapToDouble(Function<? super P_OUT, ? extends DoubleStream> mapper) {
344         Objects.requireNonNull(mapper);
345         return new DoublePipeline.StatelessOp<P_OUT>(this, StreamShape.REFERENCE,
346                                                      StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT | StreamOpFlag.NOT_SIZED) {
347             @Override
348             Sink<P_OUT> opWrapSink(int flags, Sink<Double> sink) {
349                 return new Sink.ChainedReference<P_OUT, Double>(sink) {
350                     // true if cancellationRequested() has been called
351                     boolean cancellationRequestedCalled;
352
353                     // cache the consumer to avoid creation on every accepted element
354                     DoubleConsumer downstreamAsDouble = downstream::accept;
355
356                     @Override
357                     public void begin(long size) {
358                         downstream.begin(-1);
359                     }
360
361                     @Override
362                     public void accept(P_OUT u) {
363                         try (DoubleStream result = mapper.apply(u)) {
364                             if (result != null) {
365                                 if (!cancellationRequestedCalled) {
366                                     result.sequential().forEach(downstreamAsDouble);
367                                 }
368                                 else {
369                                     var s = result.sequential().spliterator();
370                                     do { } while (!downstream.cancellationRequested() && s.tryAdvance(downstreamAsDouble));
371                                 }
372                             }
373                         }
374                     }
375
376                     @Override
377                     public boolean cancellationRequested() {
378                         cancellationRequestedCalled = true;
379                         return downstream.cancellationRequested();
380                     }
381                 };
382             }
383         };
384     }
385
386     @Override
387     public final LongStream flatMapToLong(Function<? super P_OUT, ? extends LongStream> mapper) {
388         Objects.requireNonNull(mapper);
389         // We can do better than this, by polling cancellationRequested when stream is infinite
390         return new LongPipeline.StatelessOp<P_OUT>(this, StreamShape.REFERENCE,
391                                                    StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT | StreamOpFlag.NOT_SIZED) {
392             @Override
393             Sink<P_OUT> opWrapSink(int flags, Sink<Long> sink) {
394                 return new Sink.ChainedReference<P_OUT, Long>(sink) {
395                     // true if cancellationRequested() has been called
396                     boolean cancellationRequestedCalled;
397
398                     // cache the consumer to avoid creation on every accepted element
399                     LongConsumer downstreamAsLong = downstream::accept;
400
401                     @Override
402                     public void begin(long size) {
403                         downstream.begin(-1);
404                     }
405
406                     @Override
407                     public void accept(P_OUT u) {
408                         try (LongStream result = mapper.apply(u)) {
409                             if (result != null) {
410                                 if (!cancellationRequestedCalled) {
411                                     result.sequential().forEach(downstreamAsLong);
412                                 }
413                                 else {
414                                     var s = result.sequential().spliterator();
415                                     do { } while (!downstream.cancellationRequested() && s.tryAdvance(downstreamAsLong));
416                                 }
417                             }
418                         }
419                     }
420
421                     @Override
422                     public boolean cancellationRequested() {
423                         cancellationRequestedCalled = true;
424                         return downstream.cancellationRequested();
425                     }
426                 };
427             }
428         };
429     }
430
431     @Override
432     public final Stream<P_OUT> peek(Consumer<? super P_OUT> action) {
433         Objects.requireNonNull(action);
434         return new StatelessOp<P_OUT, P_OUT>(this, StreamShape.REFERENCE,
435                                      0) {
436             @Override
437             Sink<P_OUT> opWrapSink(int flags, Sink<P_OUT> sink) {
438                 return new Sink.ChainedReference<P_OUT, P_OUT>(sink) {
439                     @Override
440                     public void accept(P_OUT u) {
441                         action.accept(u);
442                         downstream.accept(u);
443                     }
444                 };
445             }
446         };
447     }
448
449     // Stateful intermediate operations from Stream
450
451     @Override
452     public final Stream<P_OUT> distinct() {
453         return DistinctOps.makeRef(this);
454     }
455
456     @Override
457     public final Stream<P_OUT> sorted() {
458         return SortedOps.makeRef(this);
459     }
460
461     @Override
462     public final Stream<P_OUT> sorted(Comparator<? super P_OUT> comparator) {
463         return SortedOps.makeRef(this, comparator);
464     }
465
466     @Override
467     public final Stream<P_OUT> limit(long maxSize) {
468         if (maxSize < 0)
469             throw new IllegalArgumentException(Long.toString(maxSize));
470         return SliceOps.makeRef(this, 0, maxSize);
471     }
472
473     @Override
474     public final Stream<P_OUT> skip(long n) {
475         if (n < 0)
476             throw new IllegalArgumentException(Long.toString(n));
477         if (n == 0)
478             return this;
479         else
480             return SliceOps.makeRef(this, n, -1);
481     }
482
483     @Override
484     public final Stream<P_OUT> takeWhile(Predicate<? super P_OUT> predicate) {
485         return WhileOps.makeTakeWhileRef(this, predicate);
486     }
487
488     @Override
489     public final Stream<P_OUT> dropWhile(Predicate<? super P_OUT> predicate) {
490         return WhileOps.makeDropWhileRef(this, predicate);
491     }
492
493     // Terminal operations from Stream
494
495     @Override
496     public void forEach(Consumer<? super P_OUT> action) {
497         evaluate(ForEachOps.makeRef(action, false));
498     }
499
500     @Override
501     public void forEachOrdered(Consumer<? super P_OUT> action) {
502         evaluate(ForEachOps.makeRef(action, true));
503     }
504
505     @Override
506     @SuppressWarnings("unchecked")
507     public final <A> A[] toArray(IntFunction<A[]> generator) {
508         // Since A has no relation to U (not possible to declare that A is an upper bound of U)
509         // there will be no static type checking.
510         // Therefore use a raw type and assume A == U rather than propagating the separation of A and U
511         // throughout the code-base.
512         // The runtime type of U is never checked for equality with the component type of the runtime type of A[].
513         // Runtime checking will be performed when an element is stored in A[], thus if A is not a
514         // super type of U an ArrayStoreException will be thrown.
515         @SuppressWarnings("rawtypes")
516         IntFunction rawGenerator = (IntFunction) generator;
517         return (A[]) Nodes.flatten(evaluateToArrayNode(rawGenerator), rawGenerator)
518                               .asArray(rawGenerator);
519     }
520
521     @Override
522     public final Object[] toArray() {
523         return toArray(Object[]::new);
524     }
525
526     @Override
527     public final boolean anyMatch(Predicate<? super P_OUT> predicate) {
528         return evaluate(MatchOps.makeRef(predicate, MatchOps.MatchKind.ANY));
529     }
530
531     @Override
532     public final boolean allMatch(Predicate<? super P_OUT> predicate) {
533         return evaluate(MatchOps.makeRef(predicate, MatchOps.MatchKind.ALL));
534     }
535
536     @Override
537     public final boolean noneMatch(Predicate<? super P_OUT> predicate) {
538         return evaluate(MatchOps.makeRef(predicate, MatchOps.MatchKind.NONE));
539     }
540
541     @Override
542     public final Optional<P_OUT> findFirst() {
543         return evaluate(FindOps.makeRef(true));
544     }
545
546     @Override
547     public final Optional<P_OUT> findAny() {
548         return evaluate(FindOps.makeRef(false));
549     }
550
551     @Override
552     public final P_OUT reduce(final P_OUT identity, final BinaryOperator<P_OUT> accumulator) {
553         return evaluate(ReduceOps.makeRef(identity, accumulator, accumulator));
554     }
555
556     @Override
557     public final Optional<P_OUT> reduce(BinaryOperator<P_OUT> accumulator) {
558         return evaluate(ReduceOps.makeRef(accumulator));
559     }
560
561     @Override
562     public final <R> R reduce(R identity, BiFunction<R, ? super P_OUT, R> accumulator, BinaryOperator<R> combiner) {
563         return evaluate(ReduceOps.makeRef(identity, accumulator, combiner));
564     }
565
566     @Override
567     @SuppressWarnings("unchecked")
568     public final <R, A> R collect(Collector<? super P_OUT, A, R> collector) {
569         A container;
570         if (isParallel()
571                 && (collector.characteristics().contains(Collector.Characteristics.CONCURRENT))
572                 && (!isOrdered() || collector.characteristics().contains(Collector.Characteristics.UNORDERED))) {
573             container = collector.supplier().get();
574             BiConsumer<A, ? super P_OUT> accumulator = collector.accumulator();
575             forEach(u -> accumulator.accept(container, u));
576         }
577         else {
578             container = evaluate(ReduceOps.makeRef(collector));
579         }
580         return collector.characteristics().contains(Collector.Characteristics.IDENTITY_FINISH)
581                ? (R) container
582                : collector.finisher().apply(container);
583     }
584
585     @Override
586     public final <R> R collect(Supplier<R> supplier,
587                                BiConsumer<R, ? super P_OUT> accumulator,
588                                BiConsumer<R, R> combiner) {
589         return evaluate(ReduceOps.makeRef(supplier, accumulator, combiner));
590     }
591
592     @Override
593     public final Optional<P_OUT> max(Comparator<? super P_OUT> comparator) {
594         return reduce(BinaryOperator.maxBy(comparator));
595     }
596
597     @Override
598     public final Optional<P_OUT> min(Comparator<? super P_OUT> comparator) {
599         return reduce(BinaryOperator.minBy(comparator));
600
601     }
602
603     @Override
604     public final long count() {
605         return evaluate(ReduceOps.makeRefCounting());
606     }
607
608     //
609
610     /**
611      * Source stage of a ReferencePipeline.
612      *
613      * @param <E_IN> type of elements in the upstream source
614      * @param <E_OUT> type of elements in produced by this stage
615      * @since 1.8
616      */
617     static class Head<E_IN, E_OUT> extends ReferencePipeline<E_IN, E_OUT> {
618         /**
619          * Constructor for the source stage of a Stream.
620          *
621          * @param source {@code Supplier<Spliterator>} describing the stream
622          *               source
623          * @param sourceFlags the source flags for the stream source, described
624          *                    in {@link StreamOpFlag}
625          */
626         Head(Supplier<? extends Spliterator<?>> source,
627              int sourceFlags, boolean parallel) {
628             super(source, sourceFlags, parallel);
629         }
630
631         /**
632          * Constructor for the source stage of a Stream.
633          *
634          * @param source {@code Spliterator} describing the stream source
635          * @param sourceFlags the source flags for the stream source, described
636          *                    in {@link StreamOpFlag}
637          */
638         Head(Spliterator<?> source,
639              int sourceFlags, boolean parallel) {
640             super(source, sourceFlags, parallel);
641         }
642
643         @Override
644         final boolean opIsStateful() {
645             throw new UnsupportedOperationException();
646         }
647
648         @Override
649         final Sink<E_IN> opWrapSink(int flags, Sink<E_OUT> sink) {
650             throw new UnsupportedOperationException();
651         }
652
653         // Optimized sequential terminal operations for the head of the pipeline
654
655         @Override
656         public void forEach(Consumer<? super E_OUT> action) {
657             if (!isParallel()) {
658                 sourceStageSpliterator().forEachRemaining(action);
659             }
660             else {
661                 super.forEach(action);
662             }
663         }
664
665         @Override
666         public void forEachOrdered(Consumer<? super E_OUT> action) {
667             if (!isParallel()) {
668                 sourceStageSpliterator().forEachRemaining(action);
669             }
670             else {
671                 super.forEachOrdered(action);
672             }
673         }
674     }
675
676     /**
677      * Base class for a stateless intermediate stage of a Stream.
678      *
679      * @param <E_IN> type of elements in the upstream source
680      * @param <E_OUT> type of elements in produced by this stage
681      * @since 1.8
682      */
683     abstract static class StatelessOp<E_IN, E_OUT>
684             extends ReferencePipeline<E_IN, E_OUT> {
685         /**
686          * Construct a new Stream by appending a stateless intermediate
687          * operation to an existing stream.
688          *
689          * @param upstream The upstream pipeline stage
690          * @param inputShape The stream shape for the upstream pipeline stage
691          * @param opFlags Operation flags for the new stage
692          */
693         StatelessOp(AbstractPipeline<?, E_IN, ?> upstream,
694                     StreamShape inputShape,
695                     int opFlags) {
696             super(upstream, opFlags);
697             assert upstream.getOutputShape() == inputShape;
698         }
699
700         @Override
701         final boolean opIsStateful() {
702             return false;
703         }
704     }
705
706     /**
707      * Base class for a stateful intermediate stage of a Stream.
708      *
709      * @param <E_IN> type of elements in the upstream source
710      * @param <E_OUT> type of elements in produced by this stage
711      * @since 1.8
712      */
713     abstract static class StatefulOp<E_IN, E_OUT>
714             extends ReferencePipeline<E_IN, E_OUT> {
715         /**
716          * Construct a new Stream by appending a stateful intermediate operation
717          * to an existing stream.
718          * @param upstream The upstream pipeline stage
719          * @param inputShape The stream shape for the upstream pipeline stage
720          * @param opFlags Operation flags for the new stage
721          */
722         StatefulOp(AbstractPipeline<?, E_IN, ?> upstream,
723                    StreamShape inputShape,
724                    int opFlags) {
725             super(upstream, opFlags);
726             assert upstream.getOutputShape() == inputShape;
727         }
728
729         @Override
730         final boolean opIsStateful() {
731             return true;
732         }
733
734         @Override
735         abstract <P_IN> Node<E_OUT> opEvaluateParallel(PipelineHelper<E_OUT> helper,
736                                                        Spliterator<P_IN> spliterator,
737                                                        IntFunction<E_OUT[]> generator);
738     }
739 }
740