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.Comparator;
28 import java.util.Objects;
29 import java.util.Spliterator;
30 import java.util.concurrent.ConcurrentHashMap;
31 import java.util.concurrent.atomic.AtomicLong;
32 import java.util.function.BooleanSupplier;
33 import java.util.function.Consumer;
34 import java.util.function.DoubleConsumer;
35 import java.util.function.DoubleSupplier;
36 import java.util.function.IntConsumer;
37 import java.util.function.IntSupplier;
38 import java.util.function.LongConsumer;
39 import java.util.function.LongSupplier;
40 import java.util.function.Supplier;
41
42 /**
43 * Spliterator implementations for wrapping and delegating spliterators, used
44 * in the implementation of the {@link Stream#spliterator()} method.
45 *
46 * @since 1.8
47 */
48 class StreamSpliterators {
49
50 /**
51 * Abstract wrapping spliterator that binds to the spliterator of a
52 * pipeline helper on first operation.
53 *
54 * <p>This spliterator is not late-binding and will bind to the source
55 * spliterator when first operated on.
56 *
57 * <p>A wrapping spliterator produced from a sequential stream
58 * cannot be split if there are stateful operations present.
59 */
60 private abstract static class AbstractWrappingSpliterator<P_IN, P_OUT,
61 T_BUFFER extends AbstractSpinedBuffer>
62 implements Spliterator<P_OUT> {
63
64 // @@@ Detect if stateful operations are present or not
65 // If not then can split otherwise cannot
66
67 /**
68 * True if this spliterator supports splitting
69 */
70 final boolean isParallel;
71
72 final PipelineHelper<P_OUT> ph;
73
74 /**
75 * Supplier for the source spliterator. Client provides either a
76 * spliterator or a supplier.
77 */
78 private Supplier<Spliterator<P_IN>> spliteratorSupplier;
79
80 /**
81 * Source spliterator. Either provided from client or obtained from
82 * supplier.
83 */
84 Spliterator<P_IN> spliterator;
85
86 /**
87 * Sink chain for the downstream stages of the pipeline, ultimately
88 * leading to the buffer. Used during partial traversal.
89 */
90 Sink<P_IN> bufferSink;
91
92 /**
93 * A function that advances one element of the spliterator, pushing
94 * it to bufferSink. Returns whether any elements were processed.
95 * Used during partial traversal.
96 */
97 BooleanSupplier pusher;
98
99 /** Next element to consume from the buffer, used during partial traversal */
100 long nextToConsume;
101
102 /** Buffer into which elements are pushed. Used during partial traversal. */
103 T_BUFFER buffer;
104
105 /**
106 * True if full traversal has occurred (with possible cancelation).
107 * If doing a partial traversal, there may be still elements in buffer.
108 */
109 boolean finished;
110
111 /**
112 * Construct an AbstractWrappingSpliterator from a
113 * {@code Supplier<Spliterator>}.
114 */
115 AbstractWrappingSpliterator(PipelineHelper<P_OUT> ph,
116 Supplier<Spliterator<P_IN>> spliteratorSupplier,
117 boolean parallel) {
118 this.ph = ph;
119 this.spliteratorSupplier = spliteratorSupplier;
120 this.spliterator = null;
121 this.isParallel = parallel;
122 }
123
124 /**
125 * Construct an AbstractWrappingSpliterator from a
126 * {@code Spliterator}.
127 */
128 AbstractWrappingSpliterator(PipelineHelper<P_OUT> ph,
129 Spliterator<P_IN> spliterator,
130 boolean parallel) {
131 this.ph = ph;
132 this.spliteratorSupplier = null;
133 this.spliterator = spliterator;
134 this.isParallel = parallel;
135 }
136
137 /**
138 * Called before advancing to set up spliterator, if needed.
139 */
140 final void init() {
141 if (spliterator == null) {
142 spliterator = spliteratorSupplier.get();
143 spliteratorSupplier = null;
144 }
145 }
146
147 /**
148 * Get an element from the source, pushing it into the sink chain,
149 * setting up the buffer if needed
150 * @return whether there are elements to consume from the buffer
151 */
152 final boolean doAdvance() {
153 if (buffer == null) {
154 if (finished)
155 return false;
156
157 init();
158 initPartialTraversalState();
159 initPusher();
160 nextToConsume = 0;
161 bufferSink.begin(spliterator.getExactSizeIfKnown());
162 return fillBuffer();
163 }
164 else {
165 ++nextToConsume;
166 boolean hasNext = nextToConsume < buffer.count();
167 if (!hasNext) {
168 nextToConsume = 0;
169 buffer.clear();
170 hasNext = fillBuffer();
171 }
172 return hasNext;
173 }
174 }
175
176 /**
177 * Invokes the shape-specific constructor with the provided arguments
178 * and returns the result.
179 */
180 abstract AbstractWrappingSpliterator<P_IN, P_OUT, T_BUFFER> wrap(Spliterator<P_IN> s);
181
182 /**
183 * Initializes buffer, sink chain, and pusher for a shape-specific
184 * implementation.
185 */
186 abstract void initPartialTraversalState();
187
188 void initPusher() {
189 pusher = () -> spliterator.tryAdvance(bufferSink);
190 }
191
192 @Override
193 public Spliterator<P_OUT> trySplit() {
194 if (isParallel && !finished) {
195 init();
196
197 Spliterator<P_IN> split = spliterator.trySplit();
198 if (split == null)
199 return null;
200 AbstractWrappingSpliterator<P_IN, P_OUT, T_BUFFER> prefix = wrap(split);
201 if (buffer != null) {
202 prefix.buffer = buffer;
203 prefix.bufferSink = bufferSink;
204 prefix.nextToConsume = nextToConsume;
205 prefix.initPusher();
206 buffer = null;
207 }
208 return prefix;
209 }
210 else
211 return null;
212 }
213
214 /**
215 * If the buffer is empty, push elements into the sink chain until
216 * the source is empty or cancellation is requested.
217 * @return whether there are elements to consume from the buffer
218 */
219 private boolean fillBuffer() {
220 while (buffer.count() == 0) {
221 if (bufferSink.cancellationRequested() || !pusher.getAsBoolean()) {
222 if (finished)
223 return false;
224 else {
225 bufferSink.end(); // might trigger more elements
226 finished = true;
227 }
228 }
229 }
230 return true;
231 }
232
233 @Override
234 public final long estimateSize() {
235 init();
236 // Use the estimate of the wrapped spliterator
237 // Note this may not be accurate if there are filter/flatMap
238 // operations filtering or adding elements to the stream
239 return spliterator.estimateSize();
240 }
241
242 @Override
243 public final long getExactSizeIfKnown() {
244 init();
245 return StreamOpFlag.SIZED.isKnown(ph.getStreamAndOpFlags())
246 ? spliterator.getExactSizeIfKnown()
247 : -1;
248 }
249
250 @Override
251 public final int characteristics() {
252 init();
253
254 // Get the characteristics from the pipeline
255 int c = StreamOpFlag.toCharacteristics(StreamOpFlag.toStreamFlags(ph.getStreamAndOpFlags()));
256
257 // Mask off the size and uniform characteristics and replace with
258 // those of the spliterator
259 // Note that a non-uniform spliterator can change from something
260 // with an exact size to an estimate for a sub-split, for example
261 // with HashSet where the size is known at the top level spliterator
262 // but for sub-splits only an estimate is known
263 if ((c & Spliterator.SIZED) != 0) {
264 c &= ~(Spliterator.SIZED | Spliterator.SUBSIZED);
265 c |= (spliterator.characteristics() & (Spliterator.SIZED | Spliterator.SUBSIZED));
266 }
267
268 return c;
269 }
270
271 @Override
272 public Comparator<? super P_OUT> getComparator() {
273 if (!hasCharacteristics(SORTED))
274 throw new IllegalStateException();
275 return null;
276 }
277
278 @Override
279 public final String toString() {
280 return String.format("%s[%s]", getClass().getName(), spliterator);
281 }
282 }
283
284 static final class WrappingSpliterator<P_IN, P_OUT>
285 extends AbstractWrappingSpliterator<P_IN, P_OUT, SpinedBuffer<P_OUT>> {
286
287 WrappingSpliterator(PipelineHelper<P_OUT> ph,
288 Supplier<Spliterator<P_IN>> supplier,
289 boolean parallel) {
290 super(ph, supplier, parallel);
291 }
292
293 WrappingSpliterator(PipelineHelper<P_OUT> ph,
294 Spliterator<P_IN> spliterator,
295 boolean parallel) {
296 super(ph, spliterator, parallel);
297 }
298
299 @Override
300 WrappingSpliterator<P_IN, P_OUT> wrap(Spliterator<P_IN> s) {
301 return new WrappingSpliterator<>(ph, s, isParallel);
302 }
303
304 @Override
305 void initPartialTraversalState() {
306 SpinedBuffer<P_OUT> b = new SpinedBuffer<>();
307 buffer = b;
308 bufferSink = ph.wrapSink(b::accept);
309 }
310
311 @Override
312 public boolean tryAdvance(Consumer<? super P_OUT> consumer) {
313 Objects.requireNonNull(consumer);
314 boolean hasNext = doAdvance();
315 if (hasNext)
316 consumer.accept(buffer.get(nextToConsume));
317 return hasNext;
318 }
319
320 @Override
321 public void forEachRemaining(Consumer<? super P_OUT> consumer) {
322 if (buffer == null && !finished) {
323 Objects.requireNonNull(consumer);
324 init();
325
326 ph.wrapAndCopyInto((Sink<P_OUT>) consumer::accept, spliterator);
327 finished = true;
328 }
329 else {
330 do { } while (tryAdvance(consumer));
331 }
332 }
333 }
334
335 static final class IntWrappingSpliterator<P_IN>
336 extends AbstractWrappingSpliterator<P_IN, Integer, SpinedBuffer.OfInt>
337 implements Spliterator.OfInt {
338
339 IntWrappingSpliterator(PipelineHelper<Integer> ph,
340 Supplier<Spliterator<P_IN>> supplier,
341 boolean parallel) {
342 super(ph, supplier, parallel);
343 }
344
345 IntWrappingSpliterator(PipelineHelper<Integer> ph,
346 Spliterator<P_IN> spliterator,
347 boolean parallel) {
348 super(ph, spliterator, parallel);
349 }
350
351 @Override
352 AbstractWrappingSpliterator<P_IN, Integer, SpinedBuffer.OfInt> wrap(Spliterator<P_IN> s) {
353 return new IntWrappingSpliterator<>(ph, s, isParallel);
354 }
355
356 @Override
357 void initPartialTraversalState() {
358 SpinedBuffer.OfInt b = new SpinedBuffer.OfInt();
359 buffer = b;
360 bufferSink = ph.wrapSink((Sink.OfInt) b::accept);
361 }
362
363 @Override
364 public Spliterator.OfInt trySplit() {
365 return (Spliterator.OfInt) super.trySplit();
366 }
367
368 @Override
369 public boolean tryAdvance(IntConsumer consumer) {
370 Objects.requireNonNull(consumer);
371 boolean hasNext = doAdvance();
372 if (hasNext)
373 consumer.accept(buffer.get(nextToConsume));
374 return hasNext;
375 }
376
377 @Override
378 public void forEachRemaining(IntConsumer consumer) {
379 if (buffer == null && !finished) {
380 Objects.requireNonNull(consumer);
381 init();
382
383 ph.wrapAndCopyInto((Sink.OfInt) consumer::accept, spliterator);
384 finished = true;
385 }
386 else {
387 do { } while (tryAdvance(consumer));
388 }
389 }
390 }
391
392 static final class LongWrappingSpliterator<P_IN>
393 extends AbstractWrappingSpliterator<P_IN, Long, SpinedBuffer.OfLong>
394 implements Spliterator.OfLong {
395
396 LongWrappingSpliterator(PipelineHelper<Long> ph,
397 Supplier<Spliterator<P_IN>> supplier,
398 boolean parallel) {
399 super(ph, supplier, parallel);
400 }
401
402 LongWrappingSpliterator(PipelineHelper<Long> ph,
403 Spliterator<P_IN> spliterator,
404 boolean parallel) {
405 super(ph, spliterator, parallel);
406 }
407
408 @Override
409 AbstractWrappingSpliterator<P_IN, Long, SpinedBuffer.OfLong> wrap(Spliterator<P_IN> s) {
410 return new LongWrappingSpliterator<>(ph, s, isParallel);
411 }
412
413 @Override
414 void initPartialTraversalState() {
415 SpinedBuffer.OfLong b = new SpinedBuffer.OfLong();
416 buffer = b;
417 bufferSink = ph.wrapSink((Sink.OfLong) b::accept);
418 }
419
420 @Override
421 public Spliterator.OfLong trySplit() {
422 return (Spliterator.OfLong) super.trySplit();
423 }
424
425 @Override
426 public boolean tryAdvance(LongConsumer consumer) {
427 Objects.requireNonNull(consumer);
428 boolean hasNext = doAdvance();
429 if (hasNext)
430 consumer.accept(buffer.get(nextToConsume));
431 return hasNext;
432 }
433
434 @Override
435 public void forEachRemaining(LongConsumer consumer) {
436 if (buffer == null && !finished) {
437 Objects.requireNonNull(consumer);
438 init();
439
440 ph.wrapAndCopyInto((Sink.OfLong) consumer::accept, spliterator);
441 finished = true;
442 }
443 else {
444 do { } while (tryAdvance(consumer));
445 }
446 }
447 }
448
449 static final class DoubleWrappingSpliterator<P_IN>
450 extends AbstractWrappingSpliterator<P_IN, Double, SpinedBuffer.OfDouble>
451 implements Spliterator.OfDouble {
452
453 DoubleWrappingSpliterator(PipelineHelper<Double> ph,
454 Supplier<Spliterator<P_IN>> supplier,
455 boolean parallel) {
456 super(ph, supplier, parallel);
457 }
458
459 DoubleWrappingSpliterator(PipelineHelper<Double> ph,
460 Spliterator<P_IN> spliterator,
461 boolean parallel) {
462 super(ph, spliterator, parallel);
463 }
464
465 @Override
466 AbstractWrappingSpliterator<P_IN, Double, SpinedBuffer.OfDouble> wrap(Spliterator<P_IN> s) {
467 return new DoubleWrappingSpliterator<>(ph, s, isParallel);
468 }
469
470 @Override
471 void initPartialTraversalState() {
472 SpinedBuffer.OfDouble b = new SpinedBuffer.OfDouble();
473 buffer = b;
474 bufferSink = ph.wrapSink((Sink.OfDouble) b::accept);
475 }
476
477 @Override
478 public Spliterator.OfDouble trySplit() {
479 return (Spliterator.OfDouble) super.trySplit();
480 }
481
482 @Override
483 public boolean tryAdvance(DoubleConsumer consumer) {
484 Objects.requireNonNull(consumer);
485 boolean hasNext = doAdvance();
486 if (hasNext)
487 consumer.accept(buffer.get(nextToConsume));
488 return hasNext;
489 }
490
491 @Override
492 public void forEachRemaining(DoubleConsumer consumer) {
493 if (buffer == null && !finished) {
494 Objects.requireNonNull(consumer);
495 init();
496
497 ph.wrapAndCopyInto((Sink.OfDouble) consumer::accept, spliterator);
498 finished = true;
499 }
500 else {
501 do { } while (tryAdvance(consumer));
502 }
503 }
504 }
505
506 /**
507 * Spliterator implementation that delegates to an underlying spliterator,
508 * acquiring the spliterator from a {@code Supplier<Spliterator>} on the
509 * first call to any spliterator method.
510 * @param <T>
511 */
512 static class DelegatingSpliterator<T, T_SPLITR extends Spliterator<T>>
513 implements Spliterator<T> {
514 private final Supplier<? extends T_SPLITR> supplier;
515
516 private T_SPLITR s;
517
518 DelegatingSpliterator(Supplier<? extends T_SPLITR> supplier) {
519 this.supplier = supplier;
520 }
521
522 T_SPLITR get() {
523 if (s == null) {
524 s = supplier.get();
525 }
526 return s;
527 }
528
529 @Override
530 @SuppressWarnings("unchecked")
531 public T_SPLITR trySplit() {
532 return (T_SPLITR) get().trySplit();
533 }
534
535 @Override
536 public boolean tryAdvance(Consumer<? super T> consumer) {
537 return get().tryAdvance(consumer);
538 }
539
540 @Override
541 public void forEachRemaining(Consumer<? super T> consumer) {
542 get().forEachRemaining(consumer);
543 }
544
545 @Override
546 public long estimateSize() {
547 return get().estimateSize();
548 }
549
550 @Override
551 public int characteristics() {
552 return get().characteristics();
553 }
554
555 @Override
556 public Comparator<? super T> getComparator() {
557 return get().getComparator();
558 }
559
560 @Override
561 public long getExactSizeIfKnown() {
562 return get().getExactSizeIfKnown();
563 }
564
565 @Override
566 public String toString() {
567 return getClass().getName() + "[" + get() + "]";
568 }
569
570 static class OfPrimitive<T, T_CONS, T_SPLITR extends Spliterator.OfPrimitive<T, T_CONS, T_SPLITR>>
571 extends DelegatingSpliterator<T, T_SPLITR>
572 implements Spliterator.OfPrimitive<T, T_CONS, T_SPLITR> {
573 OfPrimitive(Supplier<? extends T_SPLITR> supplier) {
574 super(supplier);
575 }
576
577 @Override
578 public boolean tryAdvance(T_CONS consumer) {
579 return get().tryAdvance(consumer);
580 }
581
582 @Override
583 public void forEachRemaining(T_CONS consumer) {
584 get().forEachRemaining(consumer);
585 }
586 }
587
588 static final class OfInt
589 extends OfPrimitive<Integer, IntConsumer, Spliterator.OfInt>
590 implements Spliterator.OfInt {
591
592 OfInt(Supplier<Spliterator.OfInt> supplier) {
593 super(supplier);
594 }
595 }
596
597 static final class OfLong
598 extends OfPrimitive<Long, LongConsumer, Spliterator.OfLong>
599 implements Spliterator.OfLong {
600
601 OfLong(Supplier<Spliterator.OfLong> supplier) {
602 super(supplier);
603 }
604 }
605
606 static final class OfDouble
607 extends OfPrimitive<Double, DoubleConsumer, Spliterator.OfDouble>
608 implements Spliterator.OfDouble {
609
610 OfDouble(Supplier<Spliterator.OfDouble> supplier) {
611 super(supplier);
612 }
613 }
614 }
615
616 /**
617 * A slice Spliterator from a source Spliterator that reports
618 * {@code SUBSIZED}.
619 *
620 */
621 abstract static class SliceSpliterator<T, T_SPLITR extends Spliterator<T>> {
622 // The start index of the slice
623 final long sliceOrigin;
624 // One past the last index of the slice
625 final long sliceFence;
626
627 // The spliterator to slice
628 T_SPLITR s;
629 // current (absolute) index, modified on advance/split
630 long index;
631 // one past last (absolute) index or sliceFence, which ever is smaller
632 long fence;
633
634 SliceSpliterator(T_SPLITR s, long sliceOrigin, long sliceFence, long origin, long fence) {
635 assert s.hasCharacteristics(Spliterator.SUBSIZED);
636 this.s = s;
637 this.sliceOrigin = sliceOrigin;
638 this.sliceFence = sliceFence;
639 this.index = origin;
640 this.fence = fence;
641 }
642
643 protected abstract T_SPLITR makeSpliterator(T_SPLITR s, long sliceOrigin, long sliceFence, long origin, long fence);
644
645 public T_SPLITR trySplit() {
646 if (sliceOrigin >= fence)
647 return null;
648
649 if (index >= fence)
650 return null;
651
652 // Keep splitting until the left and right splits intersect with the slice
653 // thereby ensuring the size estimate decreases.
654 // This also avoids creating empty spliterators which can result in
655 // existing and additionally created F/J tasks that perform
656 // redundant work on no elements.
657 while (true) {
658 @SuppressWarnings("unchecked")
659 T_SPLITR leftSplit = (T_SPLITR) s.trySplit();
660 if (leftSplit == null)
661 return null;
662
663 long leftSplitFenceUnbounded = index + leftSplit.estimateSize();
664 long leftSplitFence = Math.min(leftSplitFenceUnbounded, sliceFence);
665 if (sliceOrigin >= leftSplitFence) {
666 // The left split does not intersect with, and is to the left of, the slice
667 // The right split does intersect
668 // Discard the left split and split further with the right split
669 index = leftSplitFence;
670 }
671 else if (leftSplitFence >= sliceFence) {
672 // The right split does not intersect with, and is to the right of, the slice
673 // The left split does intersect
674 // Discard the right split and split further with the left split
675 s = leftSplit;
676 fence = leftSplitFence;
677 }
678 else if (index >= sliceOrigin && leftSplitFenceUnbounded <= sliceFence) {
679 // The left split is contained within the slice, return the underlying left split
680 // Right split is contained within or intersects with the slice
681 index = leftSplitFence;
682 return leftSplit;
683 } else {
684 // The left split intersects with the slice
685 // Right split is contained within or intersects with the slice
686 return makeSpliterator(leftSplit, sliceOrigin, sliceFence, index, index = leftSplitFence);
687 }
688 }
689 }
690
691 public long estimateSize() {
692 return (sliceOrigin < fence)
693 ? fence - Math.max(sliceOrigin, index) : 0;
694 }
695
696 public int characteristics() {
697 return s.characteristics();
698 }
699
700 static final class OfRef<T>
701 extends SliceSpliterator<T, Spliterator<T>>
702 implements Spliterator<T> {
703
704 OfRef(Spliterator<T> s, long sliceOrigin, long sliceFence) {
705 this(s, sliceOrigin, sliceFence, 0, Math.min(s.estimateSize(), sliceFence));
706 }
707
708 private OfRef(Spliterator<T> s,
709 long sliceOrigin, long sliceFence, long origin, long fence) {
710 super(s, sliceOrigin, sliceFence, origin, fence);
711 }
712
713 @Override
714 protected Spliterator<T> makeSpliterator(Spliterator<T> s,
715 long sliceOrigin, long sliceFence,
716 long origin, long fence) {
717 return new OfRef<>(s, sliceOrigin, sliceFence, origin, fence);
718 }
719
720 @Override
721 public boolean tryAdvance(Consumer<? super T> action) {
722 Objects.requireNonNull(action);
723
724 if (sliceOrigin >= fence)
725 return false;
726
727 while (sliceOrigin > index) {
728 s.tryAdvance(e -> {});
729 index++;
730 }
731
732 if (index >= fence)
733 return false;
734
735 index++;
736 return s.tryAdvance(action);
737 }
738
739 @Override
740 public void forEachRemaining(Consumer<? super T> action) {
741 Objects.requireNonNull(action);
742
743 if (sliceOrigin >= fence)
744 return;
745
746 if (index >= fence)
747 return;
748
749 if (index >= sliceOrigin && (index + s.estimateSize()) <= sliceFence) {
750 // The spliterator is contained within the slice
751 s.forEachRemaining(action);
752 index = fence;
753 } else {
754 // The spliterator intersects with the slice
755 while (sliceOrigin > index) {
756 s.tryAdvance(e -> {});
757 index++;
758 }
759 // Traverse elements up to the fence
760 for (;index < fence; index++) {
761 s.tryAdvance(action);
762 }
763 }
764 }
765 }
766
767 abstract static class OfPrimitive<T,
768 T_SPLITR extends Spliterator.OfPrimitive<T, T_CONS, T_SPLITR>,
769 T_CONS>
770 extends SliceSpliterator<T, T_SPLITR>
771 implements Spliterator.OfPrimitive<T, T_CONS, T_SPLITR> {
772
773 OfPrimitive(T_SPLITR s, long sliceOrigin, long sliceFence) {
774 this(s, sliceOrigin, sliceFence, 0, Math.min(s.estimateSize(), sliceFence));
775 }
776
777 private OfPrimitive(T_SPLITR s,
778 long sliceOrigin, long sliceFence, long origin, long fence) {
779 super(s, sliceOrigin, sliceFence, origin, fence);
780 }
781
782 @Override
783 public boolean tryAdvance(T_CONS action) {
784 Objects.requireNonNull(action);
785
786 if (sliceOrigin >= fence)
787 return false;
788
789 while (sliceOrigin > index) {
790 s.tryAdvance(emptyConsumer());
791 index++;
792 }
793
794 if (index >= fence)
795 return false;
796
797 index++;
798 return s.tryAdvance(action);
799 }
800
801 @Override
802 public void forEachRemaining(T_CONS action) {
803 Objects.requireNonNull(action);
804
805 if (sliceOrigin >= fence)
806 return;
807
808 if (index >= fence)
809 return;
810
811 if (index >= sliceOrigin && (index + s.estimateSize()) <= sliceFence) {
812 // The spliterator is contained within the slice
813 s.forEachRemaining(action);
814 index = fence;
815 } else {
816 // The spliterator intersects with the slice
817 while (sliceOrigin > index) {
818 s.tryAdvance(emptyConsumer());
819 index++;
820 }
821 // Traverse elements up to the fence
822 for (;index < fence; index++) {
823 s.tryAdvance(action);
824 }
825 }
826 }
827
828 protected abstract T_CONS emptyConsumer();
829 }
830
831 static final class OfInt extends OfPrimitive<Integer, Spliterator.OfInt, IntConsumer>
832 implements Spliterator.OfInt {
833 OfInt(Spliterator.OfInt s, long sliceOrigin, long sliceFence) {
834 super(s, sliceOrigin, sliceFence);
835 }
836
837 OfInt(Spliterator.OfInt s,
838 long sliceOrigin, long sliceFence, long origin, long fence) {
839 super(s, sliceOrigin, sliceFence, origin, fence);
840 }
841
842 @Override
843 protected Spliterator.OfInt makeSpliterator(Spliterator.OfInt s,
844 long sliceOrigin, long sliceFence,
845 long origin, long fence) {
846 return new SliceSpliterator.OfInt(s, sliceOrigin, sliceFence, origin, fence);
847 }
848
849 @Override
850 protected IntConsumer emptyConsumer() {
851 return e -> {};
852 }
853 }
854
855 static final class OfLong extends OfPrimitive<Long, Spliterator.OfLong, LongConsumer>
856 implements Spliterator.OfLong {
857 OfLong(Spliterator.OfLong s, long sliceOrigin, long sliceFence) {
858 super(s, sliceOrigin, sliceFence);
859 }
860
861 OfLong(Spliterator.OfLong s,
862 long sliceOrigin, long sliceFence, long origin, long fence) {
863 super(s, sliceOrigin, sliceFence, origin, fence);
864 }
865
866 @Override
867 protected Spliterator.OfLong makeSpliterator(Spliterator.OfLong s,
868 long sliceOrigin, long sliceFence,
869 long origin, long fence) {
870 return new SliceSpliterator.OfLong(s, sliceOrigin, sliceFence, origin, fence);
871 }
872
873 @Override
874 protected LongConsumer emptyConsumer() {
875 return e -> {};
876 }
877 }
878
879 static final class OfDouble extends OfPrimitive<Double, Spliterator.OfDouble, DoubleConsumer>
880 implements Spliterator.OfDouble {
881 OfDouble(Spliterator.OfDouble s, long sliceOrigin, long sliceFence) {
882 super(s, sliceOrigin, sliceFence);
883 }
884
885 OfDouble(Spliterator.OfDouble s,
886 long sliceOrigin, long sliceFence, long origin, long fence) {
887 super(s, sliceOrigin, sliceFence, origin, fence);
888 }
889
890 @Override
891 protected Spliterator.OfDouble makeSpliterator(Spliterator.OfDouble s,
892 long sliceOrigin, long sliceFence,
893 long origin, long fence) {
894 return new SliceSpliterator.OfDouble(s, sliceOrigin, sliceFence, origin, fence);
895 }
896
897 @Override
898 protected DoubleConsumer emptyConsumer() {
899 return e -> {};
900 }
901 }
902 }
903
904 /**
905 * A slice Spliterator that does not preserve order, if any, of a source
906 * Spliterator.
907 *
908 * Note: The source spliterator may report {@code ORDERED} since that
909 * spliterator be the result of a previous pipeline stage that was
910 * collected to a {@code Node}. It is the order of the pipeline stage
911 * that governs whether the this slice spliterator is to be used or not.
912 */
913 abstract static class UnorderedSliceSpliterator<T, T_SPLITR extends Spliterator<T>> {
914 static final int CHUNK_SIZE = 1 << 7;
915
916 // The spliterator to slice
917 protected final T_SPLITR s;
918 protected final boolean unlimited;
919 private final long skipThreshold;
920 private final AtomicLong permits;
921
922 UnorderedSliceSpliterator(T_SPLITR s, long skip, long limit) {
923 this.s = s;
924 this.unlimited = limit < 0;
925 this.skipThreshold = limit >= 0 ? limit : 0;
926 this.permits = new AtomicLong(limit >= 0 ? skip + limit : skip);
927 }
928
929 UnorderedSliceSpliterator(T_SPLITR s,
930 UnorderedSliceSpliterator<T, T_SPLITR> parent) {
931 this.s = s;
932 this.unlimited = parent.unlimited;
933 this.permits = parent.permits;
934 this.skipThreshold = parent.skipThreshold;
935 }
936
937 /**
938 * Acquire permission to skip or process elements. The caller must
939 * first acquire the elements, then consult this method for guidance
940 * as to what to do with the data.
941 *
942 * <p>We use an {@code AtomicLong} to atomically maintain a counter,
943 * which is initialized as skip+limit if we are limiting, or skip only
944 * if we are not limiting. The user should consult the method
945 * {@code checkPermits()} before acquiring data elements.
946 *
947 * @param numElements the number of elements the caller has in hand
948 * @return the number of elements that should be processed; any
949 * remaining elements should be discarded.
950 */
951 protected final long acquirePermits(long numElements) {
952 long remainingPermits;
953 long grabbing;
954 // permits never increase, and don't decrease below zero
955 assert numElements > 0;
956 do {
957 remainingPermits = permits.get();
958 if (remainingPermits == 0)
959 return unlimited ? numElements : 0;
960 grabbing = Math.min(remainingPermits, numElements);
961 } while (grabbing > 0 &&
962 !permits.compareAndSet(remainingPermits, remainingPermits - grabbing));
963
964 if (unlimited)
965 return Math.max(numElements - grabbing, 0);
966 else if (remainingPermits > skipThreshold)
967 return Math.max(grabbing - (remainingPermits - skipThreshold), 0);
968 else
969 return grabbing;
970 }
971
972 enum PermitStatus { NO_MORE, MAYBE_MORE, UNLIMITED }
973
974 /** Call to check if permits might be available before acquiring data */
975 protected final PermitStatus permitStatus() {
976 if (permits.get() > 0)
977 return PermitStatus.MAYBE_MORE;
978 else
979 return unlimited ? PermitStatus.UNLIMITED : PermitStatus.NO_MORE;
980 }
981
982 public final T_SPLITR trySplit() {
983 // Stop splitting when there are no more limit permits
984 if (permits.get() == 0)
985 return null;
986 @SuppressWarnings("unchecked")
987 T_SPLITR split = (T_SPLITR) s.trySplit();
988 return split == null ? null : makeSpliterator(split);
989 }
990
991 protected abstract T_SPLITR makeSpliterator(T_SPLITR s);
992
993 public final long estimateSize() {
994 return s.estimateSize();
995 }
996
997 public final int characteristics() {
998 return s.characteristics() &
999 ~(Spliterator.SIZED | Spliterator.SUBSIZED | Spliterator.ORDERED);
1000 }
1001
1002 static final class OfRef<T> extends UnorderedSliceSpliterator<T, Spliterator<T>>
1003 implements Spliterator<T>, Consumer<T> {
1004 T tmpSlot;
1005
1006 OfRef(Spliterator<T> s, long skip, long limit) {
1007 super(s, skip, limit);
1008 }
1009
1010 OfRef(Spliterator<T> s, OfRef<T> parent) {
1011 super(s, parent);
1012 }
1013
1014 @Override
1015 public final void accept(T t) {
1016 tmpSlot = t;
1017 }
1018
1019 @Override
1020 public boolean tryAdvance(Consumer<? super T> action) {
1021 Objects.requireNonNull(action);
1022
1023 while (permitStatus() != PermitStatus.NO_MORE) {
1024 if (!s.tryAdvance(this))
1025 return false;
1026 else if (acquirePermits(1) == 1) {
1027 action.accept(tmpSlot);
1028 tmpSlot = null;
1029 return true;
1030 }
1031 }
1032 return false;
1033 }
1034
1035 @Override
1036 public void forEachRemaining(Consumer<? super T> action) {
1037 Objects.requireNonNull(action);
1038
1039 ArrayBuffer.OfRef<T> sb = null;
1040 PermitStatus permitStatus;
1041 while ((permitStatus = permitStatus()) != PermitStatus.NO_MORE) {
1042 if (permitStatus == PermitStatus.MAYBE_MORE) {
1043 // Optimistically traverse elements up to a threshold of CHUNK_SIZE
1044 if (sb == null)
1045 sb = new ArrayBuffer.OfRef<>(CHUNK_SIZE);
1046 else
1047 sb.reset();
1048 long permitsRequested = 0;
1049 do { } while (s.tryAdvance(sb) && ++permitsRequested < CHUNK_SIZE);
1050 if (permitsRequested == 0)
1051 return;
1052 sb.forEach(action, acquirePermits(permitsRequested));
1053 }
1054 else {
1055 // Must be UNLIMITED; let 'er rip
1056 s.forEachRemaining(action);
1057 return;
1058 }
1059 }
1060 }
1061
1062 @Override
1063 protected Spliterator<T> makeSpliterator(Spliterator<T> s) {
1064 return new UnorderedSliceSpliterator.OfRef<>(s, this);
1065 }
1066 }
1067
1068 /**
1069 * Concrete sub-types must also be an instance of type {@code T_CONS}.
1070 *
1071 * @param <T_BUFF> the type of the spined buffer. Must also be a type of
1072 * {@code T_CONS}.
1073 */
1074 abstract static class OfPrimitive<
1075 T,
1076 T_CONS,
1077 T_BUFF extends ArrayBuffer.OfPrimitive<T_CONS>,
1078 T_SPLITR extends Spliterator.OfPrimitive<T, T_CONS, T_SPLITR>>
1079 extends UnorderedSliceSpliterator<T, T_SPLITR>
1080 implements Spliterator.OfPrimitive<T, T_CONS, T_SPLITR> {
1081 OfPrimitive(T_SPLITR s, long skip, long limit) {
1082 super(s, skip, limit);
1083 }
1084
1085 OfPrimitive(T_SPLITR s, UnorderedSliceSpliterator.OfPrimitive<T, T_CONS, T_BUFF, T_SPLITR> parent) {
1086 super(s, parent);
1087 }
1088
1089 @Override
1090 public boolean tryAdvance(T_CONS action) {
1091 Objects.requireNonNull(action);
1092 @SuppressWarnings("unchecked")
1093 T_CONS consumer = (T_CONS) this;
1094
1095 while (permitStatus() != PermitStatus.NO_MORE) {
1096 if (!s.tryAdvance(consumer))
1097 return false;
1098 else if (acquirePermits(1) == 1) {
1099 acceptConsumed(action);
1100 return true;
1101 }
1102 }
1103 return false;
1104 }
1105
1106 protected abstract void acceptConsumed(T_CONS action);
1107
1108 @Override
1109 public void forEachRemaining(T_CONS action) {
1110 Objects.requireNonNull(action);
1111
1112 T_BUFF sb = null;
1113 PermitStatus permitStatus;
1114 while ((permitStatus = permitStatus()) != PermitStatus.NO_MORE) {
1115 if (permitStatus == PermitStatus.MAYBE_MORE) {
1116 // Optimistically traverse elements up to a threshold of CHUNK_SIZE
1117 if (sb == null)
1118 sb = bufferCreate(CHUNK_SIZE);
1119 else
1120 sb.reset();
1121 @SuppressWarnings("unchecked")
1122 T_CONS sbc = (T_CONS) sb;
1123 long permitsRequested = 0;
1124 do { } while (s.tryAdvance(sbc) && ++permitsRequested < CHUNK_SIZE);
1125 if (permitsRequested == 0)
1126 return;
1127 sb.forEach(action, acquirePermits(permitsRequested));
1128 }
1129 else {
1130 // Must be UNLIMITED; let 'er rip
1131 s.forEachRemaining(action);
1132 return;
1133 }
1134 }
1135 }
1136
1137 protected abstract T_BUFF bufferCreate(int initialCapacity);
1138 }
1139
1140 static final class OfInt
1141 extends OfPrimitive<Integer, IntConsumer, ArrayBuffer.OfInt, Spliterator.OfInt>
1142 implements Spliterator.OfInt, IntConsumer {
1143
1144 int tmpValue;
1145
1146 OfInt(Spliterator.OfInt s, long skip, long limit) {
1147 super(s, skip, limit);
1148 }
1149
1150 OfInt(Spliterator.OfInt s, UnorderedSliceSpliterator.OfInt parent) {
1151 super(s, parent);
1152 }
1153
1154 @Override
1155 public void accept(int value) {
1156 tmpValue = value;
1157 }
1158
1159 @Override
1160 protected void acceptConsumed(IntConsumer action) {
1161 action.accept(tmpValue);
1162 }
1163
1164 @Override
1165 protected ArrayBuffer.OfInt bufferCreate(int initialCapacity) {
1166 return new ArrayBuffer.OfInt(initialCapacity);
1167 }
1168
1169 @Override
1170 protected Spliterator.OfInt makeSpliterator(Spliterator.OfInt s) {
1171 return new UnorderedSliceSpliterator.OfInt(s, this);
1172 }
1173 }
1174
1175 static final class OfLong
1176 extends OfPrimitive<Long, LongConsumer, ArrayBuffer.OfLong, Spliterator.OfLong>
1177 implements Spliterator.OfLong, LongConsumer {
1178
1179 long tmpValue;
1180
1181 OfLong(Spliterator.OfLong s, long skip, long limit) {
1182 super(s, skip, limit);
1183 }
1184
1185 OfLong(Spliterator.OfLong s, UnorderedSliceSpliterator.OfLong parent) {
1186 super(s, parent);
1187 }
1188
1189 @Override
1190 public void accept(long value) {
1191 tmpValue = value;
1192 }
1193
1194 @Override
1195 protected void acceptConsumed(LongConsumer action) {
1196 action.accept(tmpValue);
1197 }
1198
1199 @Override
1200 protected ArrayBuffer.OfLong bufferCreate(int initialCapacity) {
1201 return new ArrayBuffer.OfLong(initialCapacity);
1202 }
1203
1204 @Override
1205 protected Spliterator.OfLong makeSpliterator(Spliterator.OfLong s) {
1206 return new UnorderedSliceSpliterator.OfLong(s, this);
1207 }
1208 }
1209
1210 static final class OfDouble
1211 extends OfPrimitive<Double, DoubleConsumer, ArrayBuffer.OfDouble, Spliterator.OfDouble>
1212 implements Spliterator.OfDouble, DoubleConsumer {
1213
1214 double tmpValue;
1215
1216 OfDouble(Spliterator.OfDouble s, long skip, long limit) {
1217 super(s, skip, limit);
1218 }
1219
1220 OfDouble(Spliterator.OfDouble s, UnorderedSliceSpliterator.OfDouble parent) {
1221 super(s, parent);
1222 }
1223
1224 @Override
1225 public void accept(double value) {
1226 tmpValue = value;
1227 }
1228
1229 @Override
1230 protected void acceptConsumed(DoubleConsumer action) {
1231 action.accept(tmpValue);
1232 }
1233
1234 @Override
1235 protected ArrayBuffer.OfDouble bufferCreate(int initialCapacity) {
1236 return new ArrayBuffer.OfDouble(initialCapacity);
1237 }
1238
1239 @Override
1240 protected Spliterator.OfDouble makeSpliterator(Spliterator.OfDouble s) {
1241 return new UnorderedSliceSpliterator.OfDouble(s, this);
1242 }
1243 }
1244 }
1245
1246 /**
1247 * A wrapping spliterator that only reports distinct elements of the
1248 * underlying spliterator. Does not preserve size and encounter order.
1249 */
1250 static final class DistinctSpliterator<T> implements Spliterator<T>, Consumer<T> {
1251
1252 // The value to represent null in the ConcurrentHashMap
1253 private static final Object NULL_VALUE = new Object();
1254
1255 // The underlying spliterator
1256 private final Spliterator<T> s;
1257
1258 // ConcurrentHashMap holding distinct elements as keys
1259 private final ConcurrentHashMap<T, Boolean> seen;
1260
1261 // Temporary element, only used with tryAdvance
1262 private T tmpSlot;
1263
1264 DistinctSpliterator(Spliterator<T> s) {
1265 this(s, new ConcurrentHashMap<>());
1266 }
1267
1268 private DistinctSpliterator(Spliterator<T> s, ConcurrentHashMap<T, Boolean> seen) {
1269 this.s = s;
1270 this.seen = seen;
1271 }
1272
1273 @Override
1274 public void accept(T t) {
1275 this.tmpSlot = t;
1276 }
1277
1278 @SuppressWarnings("unchecked")
1279 private T mapNull(T t) {
1280 return t != null ? t : (T) NULL_VALUE;
1281 }
1282
1283 @Override
1284 public boolean tryAdvance(Consumer<? super T> action) {
1285 while (s.tryAdvance(this)) {
1286 if (seen.putIfAbsent(mapNull(tmpSlot), Boolean.TRUE) == null) {
1287 action.accept(tmpSlot);
1288 tmpSlot = null;
1289 return true;
1290 }
1291 }
1292 return false;
1293 }
1294
1295 @Override
1296 public void forEachRemaining(Consumer<? super T> action) {
1297 s.forEachRemaining(t -> {
1298 if (seen.putIfAbsent(mapNull(t), Boolean.TRUE) == null) {
1299 action.accept(t);
1300 }
1301 });
1302 }
1303
1304 @Override
1305 public Spliterator<T> trySplit() {
1306 Spliterator<T> split = s.trySplit();
1307 return (split != null) ? new DistinctSpliterator<>(split, seen) : null;
1308 }
1309
1310 @Override
1311 public long estimateSize() {
1312 return s.estimateSize();
1313 }
1314
1315 @Override
1316 public int characteristics() {
1317 return (s.characteristics() & ~(Spliterator.SIZED | Spliterator.SUBSIZED |
1318 Spliterator.SORTED | Spliterator.ORDERED))
1319 | Spliterator.DISTINCT;
1320 }
1321
1322 @Override
1323 public Comparator<? super T> getComparator() {
1324 return s.getComparator();
1325 }
1326 }
1327
1328 /**
1329 * A Spliterator that infinitely supplies elements in no particular order.
1330 *
1331 * <p>Splitting divides the estimated size in two and stops when the
1332 * estimate size is 0.
1333 *
1334 * <p>The {@code forEachRemaining} method if invoked will never terminate.
1335 * The {@code tryAdvance} method always returns true.
1336 *
1337 */
1338 abstract static class InfiniteSupplyingSpliterator<T> implements Spliterator<T> {
1339 long estimate;
1340
1341 protected InfiniteSupplyingSpliterator(long estimate) {
1342 this.estimate = estimate;
1343 }
1344
1345 @Override
1346 public long estimateSize() {
1347 return estimate;
1348 }
1349
1350 @Override
1351 public int characteristics() {
1352 return IMMUTABLE;
1353 }
1354
1355 static final class OfRef<T> extends InfiniteSupplyingSpliterator<T> {
1356 final Supplier<T> s;
1357
1358 OfRef(long size, Supplier<T> s) {
1359 super(size);
1360 this.s = s;
1361 }
1362
1363 @Override
1364 public boolean tryAdvance(Consumer<? super T> action) {
1365 Objects.requireNonNull(action);
1366
1367 action.accept(s.get());
1368 return true;
1369 }
1370
1371 @Override
1372 public Spliterator<T> trySplit() {
1373 if (estimate == 0)
1374 return null;
1375 return new InfiniteSupplyingSpliterator.OfRef<>(estimate >>>= 1, s);
1376 }
1377 }
1378
1379 static final class OfInt extends InfiniteSupplyingSpliterator<Integer>
1380 implements Spliterator.OfInt {
1381 final IntSupplier s;
1382
1383 OfInt(long size, IntSupplier s) {
1384 super(size);
1385 this.s = s;
1386 }
1387
1388 @Override
1389 public boolean tryAdvance(IntConsumer action) {
1390 Objects.requireNonNull(action);
1391
1392 action.accept(s.getAsInt());
1393 return true;
1394 }
1395
1396 @Override
1397 public Spliterator.OfInt trySplit() {
1398 if (estimate == 0)
1399 return null;
1400 return new InfiniteSupplyingSpliterator.OfInt(estimate = estimate >>> 1, s);
1401 }
1402 }
1403
1404 static final class OfLong extends InfiniteSupplyingSpliterator<Long>
1405 implements Spliterator.OfLong {
1406 final LongSupplier s;
1407
1408 OfLong(long size, LongSupplier s) {
1409 super(size);
1410 this.s = s;
1411 }
1412
1413 @Override
1414 public boolean tryAdvance(LongConsumer action) {
1415 Objects.requireNonNull(action);
1416
1417 action.accept(s.getAsLong());
1418 return true;
1419 }
1420
1421 @Override
1422 public Spliterator.OfLong trySplit() {
1423 if (estimate == 0)
1424 return null;
1425 return new InfiniteSupplyingSpliterator.OfLong(estimate = estimate >>> 1, s);
1426 }
1427 }
1428
1429 static final class OfDouble extends InfiniteSupplyingSpliterator<Double>
1430 implements Spliterator.OfDouble {
1431 final DoubleSupplier s;
1432
1433 OfDouble(long size, DoubleSupplier s) {
1434 super(size);
1435 this.s = s;
1436 }
1437
1438 @Override
1439 public boolean tryAdvance(DoubleConsumer action) {
1440 Objects.requireNonNull(action);
1441
1442 action.accept(s.getAsDouble());
1443 return true;
1444 }
1445
1446 @Override
1447 public Spliterator.OfDouble trySplit() {
1448 if (estimate == 0)
1449 return null;
1450 return new InfiniteSupplyingSpliterator.OfDouble(estimate = estimate >>> 1, s);
1451 }
1452 }
1453 }
1454
1455 // @@@ Consolidate with Node.Builder
1456 abstract static class ArrayBuffer {
1457 int index;
1458
1459 void reset() {
1460 index = 0;
1461 }
1462
1463 static final class OfRef<T> extends ArrayBuffer implements Consumer<T> {
1464 final Object[] array;
1465
1466 OfRef(int size) {
1467 this.array = new Object[size];
1468 }
1469
1470 @Override
1471 public void accept(T t) {
1472 array[index++] = t;
1473 }
1474
1475 public void forEach(Consumer<? super T> action, long fence) {
1476 for (int i = 0; i < fence; i++) {
1477 @SuppressWarnings("unchecked")
1478 T t = (T) array[i];
1479 action.accept(t);
1480 }
1481 }
1482 }
1483
1484 abstract static class OfPrimitive<T_CONS> extends ArrayBuffer {
1485 int index;
1486
1487 @Override
1488 void reset() {
1489 index = 0;
1490 }
1491
1492 abstract void forEach(T_CONS action, long fence);
1493 }
1494
1495 static final class OfInt extends OfPrimitive<IntConsumer>
1496 implements IntConsumer {
1497 final int[] array;
1498
1499 OfInt(int size) {
1500 this.array = new int[size];
1501 }
1502
1503 @Override
1504 public void accept(int t) {
1505 array[index++] = t;
1506 }
1507
1508 @Override
1509 public void forEach(IntConsumer action, long fence) {
1510 for (int i = 0; i < fence; i++) {
1511 action.accept(array[i]);
1512 }
1513 }
1514 }
1515
1516 static final class OfLong extends OfPrimitive<LongConsumer>
1517 implements LongConsumer {
1518 final long[] array;
1519
1520 OfLong(int size) {
1521 this.array = new long[size];
1522 }
1523
1524 @Override
1525 public void accept(long t) {
1526 array[index++] = t;
1527 }
1528
1529 @Override
1530 public void forEach(LongConsumer action, long fence) {
1531 for (int i = 0; i < fence; i++) {
1532 action.accept(array[i]);
1533 }
1534 }
1535 }
1536
1537 static final class OfDouble extends OfPrimitive<DoubleConsumer>
1538 implements DoubleConsumer {
1539 final double[] array;
1540
1541 OfDouble(int size) {
1542 this.array = new double[size];
1543 }
1544
1545 @Override
1546 public void accept(double t) {
1547 array[index++] = t;
1548 }
1549
1550 @Override
1551 void forEach(DoubleConsumer action, long fence) {
1552 for (int i = 0; i < fence; i++) {
1553 action.accept(array[i]);
1554 }
1555 }
1556 }
1557 }
1558 }
1559