| /* | |
| * Copyright (c) 2023, 2025, Oracle and/or its affiliates. All rights reserved. | |
| * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER. | |
| * | |
| * This code is free software; you can redistribute it and/or modify it | |
| * under the terms of the GNU General Public License version 2 only, as | |
| * published by the Free Software Foundation. Oracle designates this | |
| * particular file as subject to the "Classpath" exception as provided | |
| * by Oracle in the LICENSE file that accompanied this code. | |
| * | |
| * This code is distributed in the hope that it will be useful, but WITHOUT | |
| * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or | |
| * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License | |
| * version 2 for more details (a copy is included in the LICENSE file that | |
| * accompanied this code). | |
| * | |
| * You should have received a copy of the GNU General Public License version | |
| * 2 along with this work; if not, write to the Free Software Foundation, | |
| * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA. | |
| * | |
| * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA | |
| * or visit www.oracle.com if you need additional information or have any | |
| * questions. | |
| */ | |
| package java.util.stream; | |
| import jdk.internal.access.SharedSecrets; | |
| import jdk.internal.vm.annotation.ForceInline; | |
| import java.util.ArrayDeque; | |
| import java.util.List; | |
| import java.util.Objects; | |
| import java.util.concurrent.Callable; | |
| import java.util.concurrent.ExecutionException; | |
| import java.util.concurrent.Future; | |
| import java.util.concurrent.FutureTask; | |
| import java.util.concurrent.Semaphore; | |
| import java.util.function.BiConsumer; | |
| import java.util.function.BiFunction; | |
| import java.util.function.BinaryOperator; | |
| import java.util.function.Function; | |
| import java.util.function.Supplier; | |
| import java.util.stream.Gatherer.Integrator; | |
| import java.util.stream.Gatherer.Downstream; | |
| /** | |
| * Implementations of {@link Gatherer} that provide useful intermediate | |
| * operations, such as windowing functions, folding functions, | |
| * transforming elements concurrently, etc. | |
| * | |
| * @since 24 | |
| */ | |
| public final class Gatherers { | |
| private Gatherers() { } // This class is not intended to be instantiated | |
| // Public built-in Gatherers and factory methods for them | |
| /** | |
| * Returns a Gatherer that gathers elements into windows | |
| * -- encounter-ordered groups of elements -- of a fixed size. | |
| * If the stream is empty then no window will be produced. | |
| * The last window may contain fewer elements than the supplied window size. | |
| * | |
| * <p>Example: | |
| * {@snippet lang = java: | |
| * // will contain: [[1, 2, 3], [4, 5, 6], [7, 8]] | |
| * List<List<Integer>> windows = | |
| * Stream.of(1,2,3,4,5,6,7,8).gather(Gatherers.windowFixed(3)).toList(); | |
| * } | |
| * | |
| * @implSpec Each window produced is an unmodifiable List; calls to any | |
| * mutator method will always cause {@code UnsupportedOperationException} | |
| * to be thrown. There are no guarantees on the implementation type or | |
| * serializability of the produced Lists. | |
| * | |
| * @apiNote For efficiency reasons, windows may be allocated contiguously | |
| * and eagerly. This means that choosing large window sizes for | |
| * small streams may use excessive memory for the duration of | |
| * evaluation of this operation. | |
| * | |
| * @param windowSize the size of the windows | |
| * @param <TR> the type of elements the returned gatherer consumes | |
| * and the contents of the windows it produces | |
| * @return a new gatherer which groups elements into fixed-size windows | |
| * @throws IllegalArgumentException when {@code windowSize} is less than 1 | |
| */ | |
| public static <TR> Gatherer<TR, ?, List<TR>> windowFixed(int windowSize) { | |
| if (windowSize < 1) | |
| throw new IllegalArgumentException("'windowSize' must be greater than zero"); | |
| class FixedWindow { | |
| Object[] window; | |
| int at; | |
| FixedWindow() { | |
| at = 0; | |
| window = new Object[windowSize]; | |
| } | |
| boolean integrate(TR element, Downstream<? super List<TR>> downstream) { | |
| window[at++] = element; | |
| if (at < windowSize) { | |
| return true; | |
| } else { | |
| final var oldWindow = window; | |
| window = new Object[windowSize]; | |
| at = 0; | |
| return downstream.push( | |
| SharedSecrets.getJavaUtilCollectionAccess() | |
| .listFromTrustedArrayNullsAllowed(oldWindow) | |
| ); | |
| } | |
| } | |
| void finish(Downstream<? super List<TR>> downstream) { | |
| if (at > 0 && !downstream.isRejecting()) { | |
| var lastWindow = new Object[at]; | |
| System.arraycopy(window, 0, lastWindow, 0, at); | |
| window = null; | |
| at = 0; | |
| downstream.push( | |
| SharedSecrets.getJavaUtilCollectionAccess() | |
| .listFromTrustedArrayNullsAllowed(lastWindow) | |
| ); | |
| } | |
| } | |
| } | |
| return Gatherer.<TR, FixedWindow, List<TR>>ofSequential( | |
| // Initializer | |
| FixedWindow::new, | |
| // Integrator | |
| Integrator.<FixedWindow, TR, List<TR>>ofGreedy(FixedWindow::integrate), | |
| // Finisher | |
| FixedWindow::finish | |
| ); | |
| } | |
| /** | |
| * Returns a Gatherer that gathers elements into windows -- | |
| * encounter-ordered groups of elements -- of a given size, where each | |
| * subsequent window includes all elements of the previous window except | |
| * for the least recent, and adds the next element in the stream. | |
| * If the stream is empty then no window will be produced. If the size of | |
| * the stream is smaller than the window size then only one window will | |
| * be produced, containing all elements in the stream. | |
| * | |
| * <p>Example: | |
| * {@snippet lang = java: | |
| * // will contain: [[1, 2], [2, 3], [3, 4], [4, 5], [5, 6], [6, 7], [7, 8]] | |
| * List<List<Integer>> windows2 = | |
| * Stream.of(1,2,3,4,5,6,7,8).gather(Gatherers.windowSliding(2)).toList(); | |
| * | |
| * // will contain: [[1, 2, 3, 4, 5, 6], [2, 3, 4, 5, 6, 7], [3, 4, 5, 6, 7, 8]] | |
| * List<List<Integer>> windows6 = | |
| * Stream.of(1,2,3,4,5,6,7,8).gather(Gatherers.windowSliding(6)).toList(); | |
| * } | |
| * | |
| * @implSpec Each window produced is an unmodifiable List; calls to any | |
| * mutator method will always cause {@code UnsupportedOperationException} | |
| * to be thrown. There are no guarantees on the implementation type or | |
| * serializability of the produced Lists. | |
| * | |
| * @apiNote For efficiency reasons, windows may be allocated contiguously | |
| * and eagerly. This means that choosing large window sizes for | |
| * small streams may use excessive memory for the duration of | |
| * evaluation of this operation. | |
| * | |
| * @param windowSize the size of the windows | |
| * @param <TR> the type of elements the returned gatherer consumes | |
| * and the contents of the windows it produces | |
| * @return a new gatherer which groups elements into sliding windows | |
| * @throws IllegalArgumentException when windowSize is less than 1 | |
| */ | |
| public static <TR> Gatherer<TR, ?, List<TR>> windowSliding(int windowSize) { | |
| if (windowSize < 1) | |
| throw new IllegalArgumentException("'windowSize' must be greater than zero"); | |
| class SlidingWindow { | |
| Object[] window; | |
| int at; | |
| boolean firstWindow; | |
| SlidingWindow() { | |
| firstWindow = true; | |
| at = 0; | |
| window = new Object[windowSize]; | |
| } | |
| boolean integrate(TR element, Downstream<? super List<TR>> downstream) { | |
| window[at++] = element; | |
| if (at < windowSize) { | |
| return true; | |
| } else { | |
| final var oldWindow = window; | |
| final var newWindow = new Object[windowSize]; | |
| System.arraycopy(oldWindow,1, newWindow, 0, windowSize - 1); | |
| window = newWindow; | |
| at -= 1; | |
| firstWindow = false; | |
| return downstream.push( | |
| SharedSecrets.getJavaUtilCollectionAccess() | |
| .listFromTrustedArrayNullsAllowed(oldWindow) | |
| ); | |
| } | |
| } | |
| void finish(Downstream<? super List<TR>> downstream) { | |
| if (firstWindow && at > 0 && !downstream.isRejecting()) { | |
| var lastWindow = new Object[at]; | |
| System.arraycopy(window, 0, lastWindow, 0, at); | |
| window = null; | |
| at = 0; | |
| downstream.push( | |
| SharedSecrets.getJavaUtilCollectionAccess() | |
| .listFromTrustedArrayNullsAllowed(lastWindow) | |
| ); | |
| } | |
| } | |
| } | |
| return Gatherer.<TR, SlidingWindow, List<TR>>ofSequential( | |
| // Initializer | |
| SlidingWindow::new, | |
| // Integrator | |
| Integrator.<SlidingWindow, TR, List<TR>>ofGreedy(SlidingWindow::integrate), | |
| // Finisher | |
| SlidingWindow::finish | |
| ); | |
| } | |
| /** | |
| * Returns a Gatherer that performs an ordered, <i>reduction-like</i>, | |
| * transformation for scenarios where no combiner-function can be | |
| * implemented, or for reductions which are intrinsically | |
| * order-dependent. | |
| * | |
| * @implSpec If no exceptions are thrown during processing, then this | |
| * operation only ever produces a single element. | |
| * | |
| * <p>Example: | |
| * {@snippet lang = java: | |
| * // will contain: Optional["123456789"] | |
| * Optional<String> numberString = | |
| * Stream.of(1,2,3,4,5,6,7,8,9) | |
| * .gather( | |
| * Gatherers.fold(() -> "", (string, number) -> string + number) | |
| * ) | |
| * .findFirst(); | |
| * } | |
| * | |
| * @see java.util.stream.Stream#reduce(Object, BinaryOperator) | |
| * | |
| * @param initial the identity value for the fold operation | |
| * @param folder the folding function | |
| * @param <T> the type of elements the returned gatherer consumes | |
| * @param <R> the type of elements the returned gatherer produces | |
| * @return a new Gatherer | |
| * @throws NullPointerException if any of the parameters are {@code null} | |
| */ | |
| public static <T, R> Gatherer<T, ?, R> fold( | |
| Supplier<R> initial, | |
| BiFunction<? super R, ? super T, ? extends R> folder) { | |
| Objects.requireNonNull(initial, "'initial' must not be null"); | |
| Objects.requireNonNull(folder, "'folder' must not be null"); | |
| class State { | |
| R value = initial.get(); | |
| State() {} | |
| } | |
| return Gatherer.ofSequential( | |
| State::new, | |
| Integrator.ofGreedy((state, element, downstream) -> { | |
| state.value = folder.apply(state.value, element); | |
| return true; | |
| }), | |
| (state, downstream) -> downstream.push(state.value) | |
| ); | |
| } | |
| /** | |
| * Returns a Gatherer that performs a Prefix Scan -- an incremental | |
| * accumulation -- using the provided functions. Starting with an | |
| * initial value obtained from the {@code Supplier}, each subsequent | |
| * value is obtained by applying the {@code BiFunction} to the current | |
| * value and the next input element, after which the resulting value is | |
| * produced downstream. | |
| * | |
| * <p>Example: | |
| * {@snippet lang = java: | |
| * // will contain: ["1", "12", "123", "1234", "12345", "123456", "1234567", "12345678", "123456789"] | |
| * List<String> numberStrings = | |
| * Stream.of(1,2,3,4,5,6,7,8,9) | |
| * .gather( | |
| * Gatherers.scan(() -> "", (string, number) -> string + number) | |
| * ) | |
| * .toList(); | |
| * } | |
| * | |
| * @param initial the supplier of the initial value for the scanner | |
| * @param scanner the function to apply for each element | |
| * @param <T> the type of element which this gatherer consumes | |
| * @param <R> the type of element which this gatherer produces | |
| * @return a new Gatherer which performs a prefix scan | |
| * @throws NullPointerException if any of the parameters are {@code null} | |
| */ | |
| public static <T, R> Gatherer<T, ?, R> scan( | |
| Supplier<R> initial, | |
| BiFunction<? super R, ? super T, ? extends R> scanner) { | |
| Objects.requireNonNull(initial, "'initial' must not be null"); | |
| Objects.requireNonNull(scanner, "'scanner' must not be null"); | |
| class State { | |
| R current = initial.get(); | |
| boolean integrate(T element, Downstream<? super R> downstream) { | |
| return downstream.push(current = scanner.apply(current, element)); | |
| } | |
| } | |
| return Gatherer.ofSequential(State::new, | |
| Integrator.<State,T, R>ofGreedy(State::integrate)); | |
| } | |
| /** | |
| * An operation which executes a function concurrently | |
| * with a configured level of max concurrency, using | |
| * <a href="{@docRoot}/java.base/java/lang/Thread.html#virtual-threads">virtual threads</a>. | |
| * This operation preserves the ordering of the stream. | |
| * | |
| * @apiNote In progress tasks will be attempted to be cancelled, | |
| * on a best-effort basis, in situations where the downstream no longer | |
| * wants to receive any more elements. | |
| * | |
| * @implSpec If a result of the function is to be pushed downstream but | |
| * instead the function completed exceptionally then the corresponding | |
| * exception will instead be rethrown by this method as an instance of | |
| * {@link RuntimeException}, after which any remaining tasks are canceled. | |
| * | |
| * @param maxConcurrency the maximum concurrency desired | |
| * @param mapper a function to be executed concurrently | |
| * @param <T> the type of input | |
| * @param <R> the type of output | |
| * @return a new Gatherer | |
| * @throws IllegalArgumentException if {@code maxConcurrency} is less than 1 | |
| * @throws NullPointerException if {@code mapper} is {@code null} | |
| */ | |
| public static <T, R> Gatherer<T,?,R> mapConcurrent( | |
| final int maxConcurrency, | |
| final Function<? super T, ? extends R> mapper) { | |
| if (maxConcurrency < 1) | |
| throw new IllegalArgumentException("'maxConcurrency' must be greater than 0"); | |
| Objects.requireNonNull(mapper, "'mapper' must not be null"); | |
| final class MapConcurrentTask extends FutureTask<R> { | |
| final Thread thread; | |
| private MapConcurrentTask(Callable<R> callable) { | |
| super(callable); | |
| this.thread = Thread.ofVirtual().unstarted(this); | |
| } | |
| } | |
| final class State { | |
| private final ArrayDeque<MapConcurrentTask> wip = | |
| new ArrayDeque<>(Math.min(maxConcurrency, 16)); | |
| boolean integrate(T element, Downstream<? super R> downstream) { | |
| // Prepare the next task and add it to the work-in-progress | |
| final var task = new MapConcurrentTask(() -> mapper.apply(element)); | |
| wip.addLast(task); | |
| assert wip.peekLast() == task; | |
| assert wip.size() <= maxConcurrency; | |
| // Start the next task | |
| task.thread.start(); | |
| // Flush at least 1 element if we're at capacity | |
| return flush(wip.size() < maxConcurrency ? 0 : 1, downstream); | |
| } | |
| boolean flush(long atLeastN, Downstream<? super R> downstream) { | |
| boolean success = false, interrupted = false; | |
| try { | |
| boolean proceed = !downstream.isRejecting(); | |
| MapConcurrentTask current; | |
| while ( | |
| proceed | |
| && (current = wip.peekFirst()) != null | |
| && (current.isDone() || atLeastN > 0) | |
| ) { | |
| R result; | |
| // Ensure that the task is done before proceeding | |
| for (;;) { | |
| try { | |
| result = current.get(); | |
| break; | |
| } catch (InterruptedException ie) { | |
| interrupted = true; // ignore for now, and restore later | |
| } | |
| } | |
| proceed &= downstream.push(result); | |
| atLeastN -= 1; | |
| final var correctRemoval = wip.pollFirst() == current; | |
| assert correctRemoval; | |
| } | |
| return (success = proceed); // Ensure that cleanup occurs if needed | |
| } catch (ExecutionException e) { | |
| final var cause = e.getCause(); | |
| throw (cause instanceof RuntimeException re) | |
| ? re | |
| : new RuntimeException(cause == null ? e : cause); | |
| } finally { | |
| // Clean up work-in-progress | |
| if (!success && !wip.isEmpty()) { | |
| // First signal cancellation for all tasks in progress | |
| for (var task : wip) | |
| task.cancel(true); | |
| // Then wait for all in progress task Threads to exit | |
| MapConcurrentTask next; | |
| while ((next = wip.pollFirst()) != null) { | |
| while (next.thread.isAlive()) { | |
| try { | |
| next.thread.join(); | |
| } catch (InterruptedException ie) { | |
| interrupted = true; // ignore, for now, and restore later | |
| } | |
| } | |
| } | |
| } | |
| // integrate(..) could be called from different threads each time | |
| // so we need to restore the interrupt on the calling thread | |
| if (interrupted) | |
| Thread.currentThread().interrupt(); | |
| } | |
| } | |
| } | |
| return Gatherer.ofSequential( | |
| State::new, | |
| Integrator.<State, T, R>ofGreedy(State::integrate), | |
| (state, downstream) -> state.flush(Long.MAX_VALUE, downstream) | |
| ); | |
| } | |
| // Implementation details | |
| /* | |
| * This enum is used to provide the default functions for the | |
| * factory methods | |
| * and for the default methods for when implementing the Gatherer interface. | |
| * | |
| * This serves the following purposes: | |
| * 1. removes the need for using `null` for signalling absence of specified | |
| * value and thereby hiding user bugs | |
| * 2. allows to check against these default values to avoid calling methods | |
| * needlessly | |
| * 3. allows for more efficient composition and evaluation | |
| */ | |
| @SuppressWarnings("rawtypes") | |
| enum Value implements Supplier, BinaryOperator, BiConsumer { | |
| DEFAULT; | |
| final BinaryOperator<Void> statelessCombiner = new BinaryOperator<>() { | |
| @Override public Void apply(Void left, Void right) { return null; } | |
| }; | |
| // BiConsumer | |
| @Override public void accept(Object state, Object downstream) {} | |
| // BinaryOperator | |
| @Override public Object apply(Object left, Object right) { | |
| throw new UnsupportedOperationException("This combiner cannot be used!"); | |
| } | |
| // Supplier | |
| @Override public Object get() { return null; } | |
| @ForceInline | |
| @SuppressWarnings("unchecked") | |
| <A> Supplier<A> initializer() { return (Supplier<A>)this; } | |
| @ForceInline | |
| @SuppressWarnings("unchecked") | |
| <T> BinaryOperator<T> combiner() { return (BinaryOperator<T>) this; } | |
| @ForceInline | |
| @SuppressWarnings("unchecked") | |
| <T, R> BiConsumer<T, Gatherer.Downstream<? super R>> finisher() { | |
| return (BiConsumer<T, Downstream<? super R>>) this; | |
| } | |
| } | |
| record GathererImpl<T, A, R>( | |
| @Override Supplier<A> initializer, | |
| @Override Integrator<A, T, R> integrator, | |
| @Override BinaryOperator<A> combiner, | |
| @Override BiConsumer<A, Downstream<? super R>> finisher) implements Gatherer<T, A, R> { | |
| static <T, A, R> GathererImpl<T, A, R> of( | |
| Supplier<A> initializer, | |
| Integrator<A, T, R> integrator, | |
| BinaryOperator<A> combiner, | |
| BiConsumer<A, Downstream<? super R>> finisher) { | |
| return new GathererImpl<>( | |
| Objects.requireNonNull(initializer,"initializer"), | |
| Objects.requireNonNull(integrator, "integrator"), | |
| Objects.requireNonNull(combiner, "combiner"), | |
| Objects.requireNonNull(finisher, "finisher") | |
| ); | |
| } | |
| } | |
| static final class Composite<T, A, R, AA, RR> implements Gatherer<T, Object, RR> { | |
| private final Gatherer<T, A, ? extends R> left; | |
| private final Gatherer<? super R, AA, ? extends RR> right; | |
| // FIXME change `impl` to a computed constant when available | |
| private GathererImpl<T, Object, RR> impl; | |
| static <T, A, R, AA, RR> Composite<T, A, R, AA, RR> of( | |
| Gatherer<T, A, ? extends R> left, | |
| Gatherer<? super R, AA, ? extends RR> right) { | |
| return new Composite<>(left, right); | |
| } | |
| private Composite(Gatherer<T, A, ? extends R> left, | |
| Gatherer<? super R, AA, ? extends RR> right) { | |
| this.left = left; | |
| this.right = right; | |
| } | |
| @SuppressWarnings("unchecked") | |
| private GathererImpl<T, Object, RR> impl() { | |
| // ATTENTION: this method currently relies on a "benign" data-race | |
| // as it should deterministically produce the same result even if | |
| // initialized concurrently on different threads. | |
| var i = impl; | |
| return i != null | |
| ? i | |
| : (impl = (GathererImpl<T, Object, RR>)impl(left, right)); | |
| } | |
| @Override public Supplier<Object> initializer() { | |
| return impl().initializer(); | |
| } | |
| @Override public Integrator<Object, T, RR> integrator() { | |
| return impl().integrator(); | |
| } | |
| @Override public BinaryOperator<Object> combiner() { | |
| return impl().combiner(); | |
| } | |
| @Override public BiConsumer<Object, Downstream<? super RR>> finisher() { | |
| return impl().finisher(); | |
| } | |
| @Override | |
| public <RRR> Gatherer<T, ?, RRR> andThen( | |
| Gatherer<? super RR, ?, ? extends RRR> that) { | |
| if (that.getClass() == Composite.class) { | |
| @SuppressWarnings("unchecked") | |
| final var c = | |
| (Composite<? super RR, ?, Object, ?, ? extends RRR>) that; | |
| return left.andThen(right.andThen(c.left).andThen(c.right)); | |
| } else { | |
| return left.andThen(right.andThen(that)); | |
| } | |
| } | |
| static final <T, A, R, AA, RR> GathererImpl<T, ?, RR> impl( | |
| Gatherer<T, A, R> left, Gatherer<? super R, AA, RR> right) { | |
| final var leftInitializer = left.initializer(); | |
| final var leftIntegrator = left.integrator(); | |
| final var leftCombiner = left.combiner(); | |
| final var leftFinisher = left.finisher(); | |
| final var rightInitializer = right.initializer(); | |
| final var rightIntegrator = right.integrator(); | |
| final var rightCombiner = right.combiner(); | |
| final var rightFinisher = right.finisher(); | |
| final var leftStateless = leftInitializer == Gatherer.defaultInitializer(); | |
| final var rightStateless = rightInitializer == Gatherer.defaultInitializer(); | |
| final var leftGreedy = leftIntegrator instanceof Integrator.Greedy; | |
| final var rightGreedy = rightIntegrator instanceof Integrator.Greedy; | |
| /* | |
| * For pairs of stateless and greedy Gatherers, we can optimize | |
| * evaluation as we do not need to track any state nor any | |
| * short-circuit signals. This can provide significant | |
| * performance improvements. | |
| */ | |
| if (leftStateless && rightStateless && leftGreedy && rightGreedy) { | |
| return new GathererImpl<>( | |
| Gatherer.defaultInitializer(), | |
| Gatherer.Integrator.ofGreedy((unused, element, downstream) -> | |
| leftIntegrator.integrate( | |
| null, | |
| element, | |
| r -> rightIntegrator.integrate(null, r, downstream)) | |
| ), | |
| (leftCombiner == Gatherer.defaultCombiner() | |
| || rightCombiner == Gatherer.defaultCombiner()) | |
| ? Gatherer.defaultCombiner() | |
| : Value.DEFAULT.statelessCombiner | |
| , | |
| (leftFinisher == Gatherer.<A,R>defaultFinisher() | |
| && rightFinisher == Gatherer.<AA,RR>defaultFinisher()) | |
| ? Gatherer.defaultFinisher() | |
| : (unused, downstream) -> { | |
| if (leftFinisher != Gatherer.<A,R>defaultFinisher()) | |
| leftFinisher.accept( | |
| null, | |
| r -> rightIntegrator.integrate(null, r, downstream)); | |
| if (rightFinisher != Gatherer.<AA,RR>defaultFinisher()) | |
| rightFinisher.accept(null, downstream); | |
| } | |
| ); | |
| } else { | |
| class State { | |
| final A leftState; | |
| final AA rightState; | |
| boolean leftProceed; | |
| boolean rightProceed; | |
| private State(A leftState, AA rightState, | |
| boolean leftProceed, boolean rightProceed) { | |
| this.leftState = leftState; | |
| this.rightState = rightState; | |
| this.leftProceed = leftProceed; | |
| this.rightProceed = rightProceed; | |
| } | |
| State() { | |
| this(leftStateless ? null : leftInitializer.get(), | |
| rightStateless ? null : rightInitializer.get(), | |
| true, true); | |
| } | |
| State joinLeft(State right) { | |
| return new State( | |
| leftStateless ? null : leftCombiner.apply(this.leftState, right.leftState), | |
| rightStateless ? null : rightCombiner.apply(this.rightState, right.rightState), | |
| this.leftProceed && this.rightProceed, | |
| right.leftProceed && right.rightProceed); | |
| } | |
| boolean integrate(T t, Downstream<? super RR> c) { | |
| /* | |
| * rightProceed must be checked after integration of | |
| * left since that can cause right to short-circuit | |
| * We always want to conditionally write leftProceed | |
| * here, which means that we only do so if we are | |
| * known to be not-greedy. | |
| */ | |
| return (leftIntegrator.integrate(leftState, t, r -> rightIntegrate(r, c)) | |
| || leftGreedy | |
| || (leftProceed = false)) | |
| && (rightGreedy || rightProceed); | |
| } | |
| void finish(Downstream<? super RR> c) { | |
| if (leftFinisher != Gatherer.<A, R>defaultFinisher()) | |
| leftFinisher.accept(leftState, r -> rightIntegrate(r, c)); | |
| if (rightFinisher != Gatherer.<AA, RR>defaultFinisher()) | |
| rightFinisher.accept(rightState, c); | |
| } | |
| /* | |
| * Currently we use the following to ferry elements from | |
| * the left Gatherer to the right Gatherer, but we create | |
| * the Gatherer.Downstream as a lambda which means that | |
| * the default implementation of `isKnownDone()` is used. | |
| * | |
| * If it is determined that we want to be able to support | |
| * the full interface of Gatherer.Downstream then we have | |
| * the following options: | |
| * 1. Have State implement Downstream<? super R> | |
| * and store the passed in Downstream<? super RR> | |
| * downstream as an instance field in integrate() | |
| * and read it in push(R r). | |
| * 2. Allocate a new Gatherer.Downstream<? super R> for | |
| * each invocation of integrate() which might prove | |
| * costly. | |
| */ | |
| public boolean rightIntegrate(R r, Downstream<? super RR> downstream) { | |
| // The following logic is highly performance sensitive | |
| return (rightGreedy || rightProceed) | |
| && (rightIntegrator.integrate(rightState, r, downstream) | |
| || rightGreedy | |
| || (rightProceed = false)); | |
| } | |
| } | |
| return new GathererImpl<T, State, RR>( | |
| State::new, | |
| (leftGreedy && rightGreedy) | |
| ? Integrator.<State, T, RR>ofGreedy(State::integrate) | |
| : Integrator.<State, T, RR>of(State::integrate), | |
| (leftCombiner == Gatherer.defaultCombiner() | |
| || rightCombiner == Gatherer.defaultCombiner()) | |
| ? Gatherer.defaultCombiner() | |
| : State::joinLeft, | |
| (leftFinisher == Gatherer.<A, R>defaultFinisher() | |
| && rightFinisher == Gatherer.<AA, RR>defaultFinisher()) | |
| ? Gatherer.defaultFinisher() | |
| : State::finish | |
| ); | |
| } | |
| } | |
| } | |
| } |