1 /*
2 * Copyright (c) 2012, 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.Objects;
28 import java.util.Spliterator;
29 import java.util.function.IntFunction;
30 import java.util.function.Supplier;
31
32 /**
33 * Abstract base class for "pipeline" classes, which are the core
34 * implementations of the Stream interface and its primitive specializations.
35 * Manages construction and evaluation of stream pipelines.
36 *
37 * <p>An {@code AbstractPipeline} represents an initial portion of a stream
38 * pipeline, encapsulating a stream source and zero or more intermediate
39 * operations. The individual {@code AbstractPipeline} objects are often
40 * referred to as <em>stages</em>, where each stage describes either the stream
41 * source or an intermediate operation.
42 *
43 * <p>A concrete intermediate stage is generally built from an
44 * {@code AbstractPipeline}, a shape-specific pipeline class which extends it
45 * (e.g., {@code IntPipeline}) which is also abstract, and an operation-specific
46 * concrete class which extends that. {@code AbstractPipeline} contains most of
47 * the mechanics of evaluating the pipeline, and implements methods that will be
48 * used by the operation; the shape-specific classes add helper methods for
49 * dealing with collection of results into the appropriate shape-specific
50 * containers.
51 *
52 * <p>After chaining a new intermediate operation, or executing a terminal
53 * operation, the stream is considered to be consumed, and no more intermediate
54 * or terminal operations are permitted on this stream instance.
55 *
56 * @implNote
57 * <p>For sequential streams, and parallel streams without
58 * <a href="package-summary.html#StreamOps">stateful intermediate
59 * operations</a>, parallel streams, pipeline evaluation is done in a single
60 * pass that "jams" all the operations together. For parallel streams with
61 * stateful operations, execution is divided into segments, where each
62 * stateful operations marks the end of a segment, and each segment is
63 * evaluated separately and the result used as the input to the next
64 * segment. In all cases, the source data is not consumed until a terminal
65 * operation begins.
66 *
67 * @param <E_IN> type of input elements
68 * @param <E_OUT> type of output elements
69 * @param <S> type of the subclass implementing {@code BaseStream}
70 * @since 1.8
71 */
72 abstract class AbstractPipeline<E_IN, E_OUT, S extends BaseStream<E_OUT, S>>
73 extends PipelineHelper<E_OUT> implements BaseStream<E_OUT, S> {
74 private static final String MSG_STREAM_LINKED = "stream has already been operated upon or closed";
75 private static final String MSG_CONSUMED = "source already consumed or closed";
76
77 /**
78 * Backlink to the head of the pipeline chain (self if this is the source
79 * stage).
80 */
81 private final AbstractPipeline sourceStage;
82
83 /**
84 * The "upstream" pipeline, or null if this is the source stage.
85 */
86 private final AbstractPipeline previousStage;
87
88 /**
89 * The operation flags for the intermediate operation represented by this
90 * pipeline object.
91 */
92 protected final int sourceOrOpFlags;
93
94 /**
95 * The next stage in the pipeline, or null if this is the last stage.
96 * Effectively final at the point of linking to the next pipeline.
97 */
98 private AbstractPipeline nextStage;
99
100 /**
101 * The number of intermediate operations between this pipeline object
102 * and the stream source if sequential, or the previous stateful if parallel.
103 * Valid at the point of pipeline preparation for evaluation.
104 */
105 private int depth;
106
107 /**
108 * The combined source and operation flags for the source and all operations
109 * up to and including the operation represented by this pipeline object.
110 * Valid at the point of pipeline preparation for evaluation.
111 */
112 private int combinedFlags;
113
114 /**
115 * The source spliterator. Only valid for the head pipeline.
116 * Before the pipeline is consumed if non-null then {@code sourceSupplier}
117 * must be null. After the pipeline is consumed if non-null then is set to
118 * null.
119 */
120 private Spliterator<?> sourceSpliterator;
121
122 /**
123 * The source supplier. Only valid for the head pipeline. Before the
124 * pipeline is consumed if non-null then {@code sourceSpliterator} must be
125 * null. After the pipeline is consumed if non-null then is set to null.
126 */
127 private Supplier<? extends Spliterator<?>> sourceSupplier;
128
129 /**
130 * True if this pipeline has been linked or consumed
131 */
132 private boolean linkedOrConsumed;
133
134 /**
135 * True if there are any stateful ops in the pipeline; only valid for the
136 * source stage.
137 */
138 private boolean sourceAnyStateful;
139
140 private Runnable sourceCloseAction;
141
142 /**
143 * True if pipeline is parallel, otherwise the pipeline is sequential; only
144 * valid for the source stage.
145 */
146 private boolean parallel;
147
148 /**
149 * Constructor for the head of a stream pipeline.
150 *
151 * @param source {@code Supplier<Spliterator>} describing the stream source
152 * @param sourceFlags The source flags for the stream source, described in
153 * {@link StreamOpFlag}
154 * @param parallel True if the pipeline is parallel
155 */
156 AbstractPipeline(Supplier<? extends Spliterator<?>> source,
157 int sourceFlags, boolean parallel) {
158 this.previousStage = null;
159 this.sourceSupplier = source;
160 this.sourceStage = this;
161 this.sourceOrOpFlags = sourceFlags & StreamOpFlag.STREAM_MASK;
162 // The following is an optimization of:
163 // StreamOpFlag.combineOpFlags(sourceOrOpFlags, StreamOpFlag.INITIAL_OPS_VALUE);
164 this.combinedFlags = (~(sourceOrOpFlags << 1)) & StreamOpFlag.INITIAL_OPS_VALUE;
165 this.depth = 0;
166 this.parallel = parallel;
167 }
168
169 /**
170 * Constructor for the head of a stream pipeline.
171 *
172 * @param source {@code Spliterator} describing the stream source
173 * @param sourceFlags the source flags for the stream source, described in
174 * {@link StreamOpFlag}
175 * @param parallel {@code true} if the pipeline is parallel
176 */
177 AbstractPipeline(Spliterator<?> source,
178 int sourceFlags, boolean parallel) {
179 this.previousStage = null;
180 this.sourceSpliterator = source;
181 this.sourceStage = this;
182 this.sourceOrOpFlags = sourceFlags & StreamOpFlag.STREAM_MASK;
183 // The following is an optimization of:
184 // StreamOpFlag.combineOpFlags(sourceOrOpFlags, StreamOpFlag.INITIAL_OPS_VALUE);
185 this.combinedFlags = (~(sourceOrOpFlags << 1)) & StreamOpFlag.INITIAL_OPS_VALUE;
186 this.depth = 0;
187 this.parallel = parallel;
188 }
189
190 /**
191 * Constructor for appending an intermediate operation stage onto an
192 * existing pipeline.
193 *
194 * @param previousStage the upstream pipeline stage
195 * @param opFlags the operation flags for the new stage, described in
196 * {@link StreamOpFlag}
197 */
198 AbstractPipeline(AbstractPipeline<?, E_IN, ?> previousStage, int opFlags) {
199 if (previousStage.linkedOrConsumed)
200 throw new IllegalStateException(MSG_STREAM_LINKED);
201 previousStage.linkedOrConsumed = true;
202 previousStage.nextStage = this;
203
204 this.previousStage = previousStage;
205 this.sourceOrOpFlags = opFlags & StreamOpFlag.OP_MASK;
206 this.combinedFlags = StreamOpFlag.combineOpFlags(opFlags, previousStage.combinedFlags);
207 this.sourceStage = previousStage.sourceStage;
208 if (opIsStateful())
209 sourceStage.sourceAnyStateful = true;
210 this.depth = previousStage.depth + 1;
211 }
212
213
214 // Terminal evaluation methods
215
216 /**
217 * Evaluate the pipeline with a terminal operation to produce a result.
218 *
219 * @param <R> the type of result
220 * @param terminalOp the terminal operation to be applied to the pipeline.
221 * @return the result
222 */
223 final <R> R evaluate(TerminalOp<E_OUT, R> terminalOp) {
224 assert getOutputShape() == terminalOp.inputShape();
225 if (linkedOrConsumed)
226 throw new IllegalStateException(MSG_STREAM_LINKED);
227 linkedOrConsumed = true;
228
229 return isParallel()
230 ? (R) terminalOp.evaluateParallel(this, sourceSpliterator(terminalOp.getOpFlags()))
231 : (R) terminalOp.evaluateSequential(this, sourceSpliterator(terminalOp.getOpFlags()));
232 }
233
234 /**
235 * Collect the elements output from the pipeline stage.
236 *
237 * @param generator the array generator to be used to create array instances
238 * @return a flat array-backed Node that holds the collected output elements
239 */
240 final Node<E_OUT> evaluateToArrayNode(IntFunction<E_OUT[]> generator) {
241 if (linkedOrConsumed)
242 throw new IllegalStateException(MSG_STREAM_LINKED);
243 linkedOrConsumed = true;
244
245 // If the last intermediate operation is stateful then
246 // evaluate directly to avoid an extra collection step
247 if (isParallel() && previousStage != null && opIsStateful()) {
248 return opEvaluateParallel(previousStage, previousStage.sourceSpliterator(0), generator);
249 }
250 else {
251 return evaluate(sourceSpliterator(0), true, generator);
252 }
253 }
254
255 /**
256 * Gets the source stage spliterator if this pipeline stage is the source
257 * stage. The pipeline is consumed after this method is called and
258 * returns successfully.
259 *
260 * @return the source stage spliterator
261 * @throws IllegalStateException if this pipeline stage is not the source
262 * stage.
263 */
264 final Spliterator<E_OUT> sourceStageSpliterator() {
265 if (this != sourceStage)
266 throw new IllegalStateException();
267
268 if (linkedOrConsumed)
269 throw new IllegalStateException(MSG_STREAM_LINKED);
270 linkedOrConsumed = true;
271
272 if (sourceStage.sourceSpliterator != null) {
273 Spliterator<E_OUT> s = sourceStage.sourceSpliterator;
274 sourceStage.sourceSpliterator = null;
275 return s;
276 }
277 else if (sourceStage.sourceSupplier != null) {
278 Spliterator<E_OUT> s = (Spliterator<E_OUT>) sourceStage.sourceSupplier.get();
279 sourceStage.sourceSupplier = null;
280 return s;
281 }
282 else {
283 throw new IllegalStateException(MSG_CONSUMED);
284 }
285 }
286
287 // BaseStream
288
289 @Override
290 public final S sequential() {
291 sourceStage.parallel = false;
292 return (S) this;
293 }
294
295 @Override
296 public final S parallel() {
297 sourceStage.parallel = true;
298 return (S) this;
299 }
300
301 @Override
302 public void close() {
303 linkedOrConsumed = true;
304 sourceSupplier = null;
305 sourceSpliterator = null;
306 if (sourceStage.sourceCloseAction != null) {
307 sourceStage.sourceCloseAction.run();
308 sourceStage.sourceCloseAction = null;
309 }
310 }
311
312 @Override
313 public S onClose(Runnable closeHandler) {
314 Runnable existingHandler = sourceStage.sourceCloseAction;
315 sourceStage.sourceCloseAction =
316 (existingHandler == null)
317 ? closeHandler
318 : Streams.composeWithExceptions(existingHandler, closeHandler);
319 return (S) this;
320 }
321
322 // Primitive specialization use co-variant overrides, hence is not final
323 @Override
324 public Spliterator<E_OUT> spliterator() {
325 if (linkedOrConsumed)
326 throw new IllegalStateException(MSG_STREAM_LINKED);
327 linkedOrConsumed = true;
328
329 if (this == sourceStage) {
330 if (sourceStage.sourceSpliterator != null) {
331 Spliterator<E_OUT> s = sourceStage.sourceSpliterator;
332 sourceStage.sourceSpliterator = null;
333 return s;
334 }
335 else if (sourceStage.sourceSupplier != null) {
336 Supplier<Spliterator<E_OUT>> s = sourceStage.sourceSupplier;
337 sourceStage.sourceSupplier = null;
338 return lazySpliterator(s);
339 }
340 else {
341 throw new IllegalStateException(MSG_CONSUMED);
342 }
343 }
344 else {
345 return wrap(this, () -> sourceSpliterator(0), isParallel());
346 }
347 }
348
349 @Override
350 public final boolean isParallel() {
351 return sourceStage.parallel;
352 }
353
354
355 /**
356 * Returns the composition of stream flags of the stream source and all
357 * intermediate operations.
358 *
359 * @return the composition of stream flags of the stream source and all
360 * intermediate operations
361 * @see StreamOpFlag
362 */
363 final int getStreamFlags() {
364 return StreamOpFlag.toStreamFlags(combinedFlags);
365 }
366
367 /**
368 * Prepare the pipeline for a parallel execution. As the pipeline is built,
369 * the flags and depth indicators are set up for a sequential execution.
370 * If the execution is parallel, and there are any stateful operations, then
371 * some of these need to be adjusted, as well as adjusting for flags from
372 * the terminal operation (such as back-propagating UNORDERED).
373 * Need not be called for a sequential execution.
374 *
375 * @param terminalFlags Operation flags for the terminal operation
376 */
377 private void parallelPrepare(int terminalFlags) {
378 AbstractPipeline backPropagationHead = sourceStage;
379 if (sourceStage.sourceAnyStateful) {
380 int depth = 1;
381 for (AbstractPipeline u = sourceStage, p = sourceStage.nextStage;
382 p != null;
383 u = p, p = p.nextStage) {
384 int thisOpFlags = p.sourceOrOpFlags;
385 if (p.opIsStateful()) {
386 // If the stateful operation is a short-circuit operation
387 // then move the back propagation head forwards
388 // NOTE: there are no size-injecting ops
389 if (StreamOpFlag.SHORT_CIRCUIT.isKnown(thisOpFlags)) {
390 backPropagationHead = p;
391 // Clear the short circuit flag for next pipeline stage
392 // This stage encapsulates short-circuiting, the next
393 // stage may not have any short-circuit operations, and
394 // if so spliterator.forEachRemaining should be be used
395 // for traversal
396 thisOpFlags = thisOpFlags & ~StreamOpFlag.IS_SHORT_CIRCUIT;
397 }
398
399 depth = 0;
400 // The following injects size, it is equivalent to:
401 // StreamOpFlag.combineOpFlags(StreamOpFlag.IS_SIZED, p.combinedFlags);
402 thisOpFlags = (thisOpFlags & ~StreamOpFlag.NOT_SIZED) | StreamOpFlag.IS_SIZED;
403 }
404 p.depth = depth++;
405 p.combinedFlags = StreamOpFlag.combineOpFlags(thisOpFlags, u.combinedFlags);
406 }
407 }
408
409 // Apply the upstream terminal flags
410 if (terminalFlags != 0) {
411 int upstreamTerminalFlags = terminalFlags & StreamOpFlag.UPSTREAM_TERMINAL_OP_MASK;
412 for (AbstractPipeline p = backPropagationHead; p.nextStage != null; p = p.nextStage) {
413 p.combinedFlags = StreamOpFlag.combineOpFlags(upstreamTerminalFlags, p.combinedFlags);
414 }
415
416 combinedFlags = StreamOpFlag.combineOpFlags(terminalFlags, combinedFlags);
417 }
418 }
419
420 /**
421 * Get the source spliterator for this pipeline stage. For a sequential or
422 * stateless parallel pipeline, this is the source spliterator. For a
423 * stateful parallel pipeline, this is a spliterator describing the results
424 * of all computations up to and including the most recent stateful
425 * operation.
426 */
427 private Spliterator<?> sourceSpliterator(int terminalFlags) {
428 // Get the source spliterator of the pipeline
429 Spliterator<?> spliterator = null;
430 if (sourceStage.sourceSpliterator != null) {
431 spliterator = sourceStage.sourceSpliterator;
432 sourceStage.sourceSpliterator = null;
433 }
434 else if (sourceStage.sourceSupplier != null) {
435 spliterator = (Spliterator<?>) sourceStage.sourceSupplier.get();
436 sourceStage.sourceSupplier = null;
437 }
438 else {
439 throw new IllegalStateException(MSG_CONSUMED);
440 }
441
442 if (isParallel()) {
443 // @@@ Merge parallelPrepare with the loop below and use the
444 // spliterator characteristics to determine if SIZED
445 // should be injected
446 parallelPrepare(terminalFlags);
447
448 // Adapt the source spliterator, evaluating each stateful op
449 // in the pipeline up to and including this pipeline stage
450 for (AbstractPipeline u = sourceStage, p = sourceStage.nextStage, e = this;
451 u != e;
452 u = p, p = p.nextStage) {
453
454 if (p.opIsStateful()) {
455 spliterator = p.opEvaluateParallelLazy(u, spliterator);
456 }
457 }
458 }
459 else if (terminalFlags != 0) {
460 combinedFlags = StreamOpFlag.combineOpFlags(terminalFlags, combinedFlags);
461 }
462
463 return spliterator;
464 }
465
466
467 // PipelineHelper
468
469 @Override
470 final StreamShape getSourceShape() {
471 AbstractPipeline p = AbstractPipeline.this;
472 while (p.depth > 0) {
473 p = p.previousStage;
474 }
475 return p.getOutputShape();
476 }
477
478 @Override
479 final <P_IN> long exactOutputSizeIfKnown(Spliterator<P_IN> spliterator) {
480 return StreamOpFlag.SIZED.isKnown(getStreamAndOpFlags()) ? spliterator.getExactSizeIfKnown() : -1;
481 }
482
483 @Override
484 final <P_IN, S extends Sink<E_OUT>> S wrapAndCopyInto(S sink, Spliterator<P_IN> spliterator) {
485 copyInto(wrapSink(Objects.requireNonNull(sink)), spliterator);
486 return sink;
487 }
488
489 @Override
490 final <P_IN> void copyInto(Sink<P_IN> wrappedSink, Spliterator<P_IN> spliterator) {
491 Objects.requireNonNull(wrappedSink);
492
493 if (!StreamOpFlag.SHORT_CIRCUIT.isKnown(getStreamAndOpFlags())) {
494 wrappedSink.begin(spliterator.getExactSizeIfKnown());
495 spliterator.forEachRemaining(wrappedSink);
496 wrappedSink.end();
497 }
498 else {
499 copyIntoWithCancel(wrappedSink, spliterator);
500 }
501 }
502
503 @Override
504 final <P_IN> void copyIntoWithCancel(Sink<P_IN> wrappedSink, Spliterator<P_IN> spliterator) {
505 AbstractPipeline p = AbstractPipeline.this;
506 while (p.depth > 0) {
507 p = p.previousStage;
508 }
509 wrappedSink.begin(spliterator.getExactSizeIfKnown());
510 p.forEachWithCancel(spliterator, wrappedSink);
511 wrappedSink.end();
512 }
513
514 @Override
515 final int getStreamAndOpFlags() {
516 return combinedFlags;
517 }
518
519 final boolean isOrdered() {
520 return StreamOpFlag.ORDERED.isKnown(combinedFlags);
521 }
522
523 @Override
524 final <P_IN> Sink<P_IN> wrapSink(Sink<E_OUT> sink) {
525 Objects.requireNonNull(sink);
526
527 for (AbstractPipeline p=AbstractPipeline.this; p.depth > 0; p=p.previousStage) {
528 sink = p.opWrapSink(p.previousStage.combinedFlags, sink);
529 }
530 return (Sink<P_IN>) sink;
531 }
532
533 @Override
534 final <P_IN> Spliterator<E_OUT> wrapSpliterator(Spliterator<P_IN> sourceSpliterator) {
535 if (depth == 0) {
536 return (Spliterator<E_OUT>) sourceSpliterator;
537 }
538 else {
539 return wrap(this, () -> sourceSpliterator, isParallel());
540 }
541 }
542
543 @Override
544 @SuppressWarnings("unchecked")
545 final <P_IN> Node<E_OUT> evaluate(Spliterator<P_IN> spliterator,
546 boolean flatten,
547 IntFunction<E_OUT[]> generator) {
548 if (isParallel()) {
549 // @@@ Optimize if op of this pipeline stage is a stateful op
550 return evaluateToNode(this, spliterator, flatten, generator);
551 }
552 else {
553 Node.Builder<E_OUT> nb = makeNodeBuilder(
554 exactOutputSizeIfKnown(spliterator), generator);
555 return wrapAndCopyInto(nb, spliterator).build();
556 }
557 }
558
559
560 // Shape-specific abstract methods, implemented by XxxPipeline classes
561
562 /**
563 * Get the output shape of the pipeline. If the pipeline is the head,
564 * then it's output shape corresponds to the shape of the source.
565 * Otherwise, it's output shape corresponds to the output shape of the
566 * associated operation.
567 *
568 * @return the output shape
569 */
570 abstract StreamShape getOutputShape();
571
572 /**
573 * Collect elements output from a pipeline into a Node that holds elements
574 * of this shape.
575 *
576 * @param helper the pipeline helper describing the pipeline stages
577 * @param spliterator the source spliterator
578 * @param flattenTree true if the returned node should be flattened
579 * @param generator the array generator
580 * @return a Node holding the output of the pipeline
581 */
582 abstract <P_IN> Node<E_OUT> evaluateToNode(PipelineHelper<E_OUT> helper,
583 Spliterator<P_IN> spliterator,
584 boolean flattenTree,
585 IntFunction<E_OUT[]> generator);
586
587 /**
588 * Create a spliterator that wraps a source spliterator, compatible with
589 * this stream shape, and operations associated with a {@link
590 * PipelineHelper}.
591 *
592 * @param ph the pipeline helper describing the pipeline stages
593 * @param supplier the supplier of a spliterator
594 * @return a wrapping spliterator compatible with this shape
595 */
596 abstract <P_IN> Spliterator<E_OUT> wrap(PipelineHelper<E_OUT> ph,
597 Supplier<Spliterator<P_IN>> supplier,
598 boolean isParallel);
599
600 /**
601 * Create a lazy spliterator that wraps and obtains the supplied the
602 * spliterator when a method is invoked on the lazy spliterator.
603 * @param supplier the supplier of a spliterator
604 */
605 abstract Spliterator<E_OUT> lazySpliterator(Supplier<? extends Spliterator<E_OUT>> supplier);
606
607 /**
608 * Traverse the elements of a spliterator compatible with this stream shape,
609 * pushing those elements into a sink. If the sink requests cancellation,
610 * no further elements will be pulled or pushed.
611 *
612 * @param spliterator the spliterator to pull elements from
613 * @param sink the sink to push elements to
614 */
615 abstract void forEachWithCancel(Spliterator<E_OUT> spliterator, Sink<E_OUT> sink);
616
617 /**
618 * Make a node builder compatible with this stream shape.
619 *
620 * @param exactSizeIfKnown if {@literal >=0}, then a node builder will be created that
621 * has a fixed capacity of at most sizeIfKnown elements. If {@literal < 0},
622 * then the node builder has an unfixed capacity. A fixed capacity node
623 * builder will throw exceptions if an element is added after builder has
624 * reached capacity, or is built before the builder has reached capacity.
625 * @param generator the array generator to be used to create instances of a
626 * T[] array. For implementations supporting primitive nodes, this parameter
627 * may be ignored.
628 * @return a node builder
629 */
630 abstract Node.Builder<E_OUT> makeNodeBuilder(long exactSizeIfKnown,
631 IntFunction<E_OUT[]> generator);
632
633
634 // Op-specific abstract methods, implemented by the operation class
635
636 /**
637 * Returns whether this operation is stateful or not. If it is stateful,
638 * then the method
639 * {@link #opEvaluateParallel(PipelineHelper, java.util.Spliterator, java.util.function.IntFunction)}
640 * must be overridden.
641 *
642 * @return {@code true} if this operation is stateful
643 */
644 abstract boolean opIsStateful();
645
646 /**
647 * Accepts a {@code Sink} which will receive the results of this operation,
648 * and return a {@code Sink} which accepts elements of the input type of
649 * this operation and which performs the operation, passing the results to
650 * the provided {@code Sink}.
651 *
652 * @apiNote
653 * The implementation may use the {@code flags} parameter to optimize the
654 * sink wrapping. For example, if the input is already {@code DISTINCT},
655 * the implementation for the {@code Stream#distinct()} method could just
656 * return the sink it was passed.
657 *
658 * @param flags The combined stream and operation flags up to, but not
659 * including, this operation
660 * @param sink sink to which elements should be sent after processing
661 * @return a sink which accepts elements, perform the operation upon
662 * each element, and passes the results (if any) to the provided
663 * {@code Sink}.
664 */
665 abstract Sink<E_IN> opWrapSink(int flags, Sink<E_OUT> sink);
666
667 /**
668 * Performs a parallel evaluation of the operation using the specified
669 * {@code PipelineHelper} which describes the upstream intermediate
670 * operations. Only called on stateful operations. If {@link
671 * #opIsStateful()} returns true then implementations must override the
672 * default implementation.
673 *
674 * @implSpec The default implementation always throw
675 * {@code UnsupportedOperationException}.
676 *
677 * @param helper the pipeline helper describing the pipeline stages
678 * @param spliterator the source {@code Spliterator}
679 * @param generator the array generator
680 * @return a {@code Node} describing the result of the evaluation
681 */
682 <P_IN> Node<E_OUT> opEvaluateParallel(PipelineHelper<E_OUT> helper,
683 Spliterator<P_IN> spliterator,
684 IntFunction<E_OUT[]> generator) {
685 throw new UnsupportedOperationException("Parallel evaluation is not supported");
686 }
687
688 /**
689 * Returns a {@code Spliterator} describing a parallel evaluation of the
690 * operation, using the specified {@code PipelineHelper} which describes the
691 * upstream intermediate operations. Only called on stateful operations.
692 * It is not necessary (though acceptable) to do a full computation of the
693 * result here; it is preferable, if possible, to describe the result via a
694 * lazily evaluated spliterator.
695 *
696 * @implSpec The default implementation behaves as if:
697 * <pre>{@code
698 * return evaluateParallel(helper, i -> (E_OUT[]) new
699 * Object[i]).spliterator();
700 * }</pre>
701 * and is suitable for implementations that cannot do better than a full
702 * synchronous evaluation.
703 *
704 * @param helper the pipeline helper
705 * @param spliterator the source {@code Spliterator}
706 * @return a {@code Spliterator} describing the result of the evaluation
707 */
708 <P_IN> Spliterator<E_OUT> opEvaluateParallelLazy(PipelineHelper<E_OUT> helper,
709 Spliterator<P_IN> spliterator) {
710 return opEvaluateParallel(helper, spliterator, i -> (E_OUT[]) new Object[i]).spliterator();
711 }
712 }