365 lines
14 KiB
Java
365 lines
14 KiB
Java
/*
|
|
* Copyright (c) 2012, 2013, 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 java.util.Objects;
|
|
import java.util.function.Consumer;
|
|
import java.util.function.DoubleConsumer;
|
|
import java.util.function.IntConsumer;
|
|
import java.util.function.LongConsumer;
|
|
|
|
/**
|
|
* An extension of {@link Consumer} used to conduct values through the stages of
|
|
* a stream pipeline, with additional methods to manage size information,
|
|
* control flow, etc. Before calling the {@code accept()} method on a
|
|
* {@code Sink} for the first time, you must first call the {@code begin()}
|
|
* method to inform it that data is coming (optionally informing the sink how
|
|
* much data is coming), and after all data has been sent, you must call the
|
|
* {@code end()} method. After calling {@code end()}, you should not call
|
|
* {@code accept()} without again calling {@code begin()}. {@code Sink} also
|
|
* offers a mechanism by which the sink can cooperatively signal that it does
|
|
* not wish to receive any more data (the {@code cancellationRequested()}
|
|
* method), which a source can poll before sending more data to the
|
|
* {@code Sink}.
|
|
*
|
|
* <p>A sink may be in one of two states: an initial state and an active state.
|
|
* It starts out in the initial state; the {@code begin()} method transitions
|
|
* it to the active state, and the {@code end()} method transitions it back into
|
|
* the initial state, where it can be re-used. Data-accepting methods (such as
|
|
* {@code accept()} are only valid in the active state.
|
|
*
|
|
* @apiNote
|
|
* A stream pipeline consists of a source, zero or more intermediate stages
|
|
* (such as filtering or mapping), and a terminal stage, such as reduction or
|
|
* for-each. For concreteness, consider the pipeline:
|
|
*
|
|
* <pre>{@code
|
|
* int longestStringLengthStartingWithA
|
|
* = strings.stream()
|
|
* .filter(s -> s.startsWith("A"))
|
|
* .mapToInt(String::length)
|
|
* .max();
|
|
* }</pre>
|
|
*
|
|
* <p>Here, we have three stages, filtering, mapping, and reducing. The
|
|
* filtering stage consumes strings and emits a subset of those strings; the
|
|
* mapping stage consumes strings and emits ints; the reduction stage consumes
|
|
* those ints and computes the maximal value.
|
|
*
|
|
* <p>A {@code Sink} instance is used to represent each stage of this pipeline,
|
|
* whether the stage accepts objects, ints, longs, or doubles. Sink has entry
|
|
* points for {@code accept(Object)}, {@code accept(int)}, etc, so that we do
|
|
* not need a specialized interface for each primitive specialization. (It
|
|
* might be called a "kitchen sink" for this omnivorous tendency.) The entry
|
|
* point to the pipeline is the {@code Sink} for the filtering stage, which
|
|
* sends some elements "downstream" -- into the {@code Sink} for the mapping
|
|
* stage, which in turn sends integral values downstream into the {@code Sink}
|
|
* for the reduction stage. The {@code Sink} implementations associated with a
|
|
* given stage is expected to know the data type for the next stage, and call
|
|
* the correct {@code accept} method on its downstream {@code Sink}. Similarly,
|
|
* each stage must implement the correct {@code accept} method corresponding to
|
|
* the data type it accepts.
|
|
*
|
|
* <p>The specialized subtypes such as {@link Sink.OfInt} override
|
|
* {@code accept(Object)} to call the appropriate primitive specialization of
|
|
* {@code accept}, implement the appropriate primitive specialization of
|
|
* {@code Consumer}, and re-abstract the appropriate primitive specialization of
|
|
* {@code accept}.
|
|
*
|
|
* <p>The chaining subtypes such as {@link ChainedInt} not only implement
|
|
* {@code Sink.OfInt}, but also maintain a {@code downstream} field which
|
|
* represents the downstream {@code Sink}, and implement the methods
|
|
* {@code begin()}, {@code end()}, and {@code cancellationRequested()} to
|
|
* delegate to the downstream {@code Sink}. Most implementations of
|
|
* intermediate operations will use these chaining wrappers. For example, the
|
|
* mapping stage in the above example would look like:
|
|
*
|
|
* <pre>{@code
|
|
* IntSink is = new Sink.ChainedReference<U>(sink) {
|
|
* public void accept(U u) {
|
|
* downstream.accept(mapper.applyAsInt(u));
|
|
* }
|
|
* };
|
|
* }</pre>
|
|
*
|
|
* <p>Here, we implement {@code Sink.ChainedReference<U>}, meaning that we expect
|
|
* to receive elements of type {@code U} as input, and pass the downstream sink
|
|
* to the constructor. Because the next stage expects to receive integers, we
|
|
* must call the {@code accept(int)} method when emitting values to the downstream.
|
|
* The {@code accept()} method applies the mapping function from {@code U} to
|
|
* {@code int} and passes the resulting value to the downstream {@code Sink}.
|
|
*
|
|
* @param <T> type of elements for value streams
|
|
* @since 1.8
|
|
* @hide Visible for CTS testing only (OpenJDK8 tests).
|
|
*/
|
|
// Android-changed: Made public for CTS tests only.
|
|
public interface Sink<T> extends Consumer<T> {
|
|
/**
|
|
* Resets the sink state to receive a fresh data set. This must be called
|
|
* before sending any data to the sink. After calling {@link #end()},
|
|
* you may call this method to reset the sink for another calculation.
|
|
* @param size The exact size of the data to be pushed downstream, if
|
|
* known or {@code -1} if unknown or infinite.
|
|
*
|
|
* <p>Prior to this call, the sink must be in the initial state, and after
|
|
* this call it is in the active state.
|
|
*/
|
|
default void begin(long size) {}
|
|
|
|
/**
|
|
* Indicates that all elements have been pushed. If the {@code Sink} is
|
|
* stateful, it should send any stored state downstream at this time, and
|
|
* should clear any accumulated state (and associated resources).
|
|
*
|
|
* <p>Prior to this call, the sink must be in the active state, and after
|
|
* this call it is returned to the initial state.
|
|
*/
|
|
default void end() {}
|
|
|
|
/**
|
|
* Indicates that this {@code Sink} does not wish to receive any more data.
|
|
*
|
|
* @implSpec The default implementation always returns false.
|
|
*
|
|
* @return true if cancellation is requested
|
|
*/
|
|
default boolean cancellationRequested() {
|
|
return false;
|
|
}
|
|
|
|
/**
|
|
* Accepts an int value.
|
|
*
|
|
* @implSpec The default implementation throws IllegalStateException.
|
|
*
|
|
* @throws IllegalStateException if this sink does not accept int values
|
|
*/
|
|
default void accept(int value) {
|
|
throw new IllegalStateException("called wrong accept method");
|
|
}
|
|
|
|
/**
|
|
* Accepts a long value.
|
|
*
|
|
* @implSpec The default implementation throws IllegalStateException.
|
|
*
|
|
* @throws IllegalStateException if this sink does not accept long values
|
|
*/
|
|
default void accept(long value) {
|
|
throw new IllegalStateException("called wrong accept method");
|
|
}
|
|
|
|
/**
|
|
* Accepts a double value.
|
|
*
|
|
* @implSpec The default implementation throws IllegalStateException.
|
|
*
|
|
* @throws IllegalStateException if this sink does not accept double values
|
|
*/
|
|
default void accept(double value) {
|
|
throw new IllegalStateException("called wrong accept method");
|
|
}
|
|
|
|
/**
|
|
* {@code Sink} that implements {@code Sink<Integer>}, re-abstracts
|
|
* {@code accept(int)}, and wires {@code accept(Integer)} to bridge to
|
|
* {@code accept(int)}.
|
|
*/
|
|
interface OfInt extends Sink<Integer>, IntConsumer {
|
|
@Override
|
|
void accept(int value);
|
|
|
|
@Override
|
|
default void accept(Integer i) {
|
|
if (Tripwire.ENABLED)
|
|
Tripwire.trip(getClass(), "{0} calling Sink.OfInt.accept(Integer)");
|
|
accept(i.intValue());
|
|
}
|
|
}
|
|
|
|
/**
|
|
* {@code Sink} that implements {@code Sink<Long>}, re-abstracts
|
|
* {@code accept(long)}, and wires {@code accept(Long)} to bridge to
|
|
* {@code accept(long)}.
|
|
*/
|
|
interface OfLong extends Sink<Long>, LongConsumer {
|
|
@Override
|
|
void accept(long value);
|
|
|
|
@Override
|
|
default void accept(Long i) {
|
|
if (Tripwire.ENABLED)
|
|
Tripwire.trip(getClass(), "{0} calling Sink.OfLong.accept(Long)");
|
|
accept(i.longValue());
|
|
}
|
|
}
|
|
|
|
/**
|
|
* {@code Sink} that implements {@code Sink<Double>}, re-abstracts
|
|
* {@code accept(double)}, and wires {@code accept(Double)} to bridge to
|
|
* {@code accept(double)}.
|
|
*/
|
|
interface OfDouble extends Sink<Double>, DoubleConsumer {
|
|
@Override
|
|
void accept(double value);
|
|
|
|
@Override
|
|
default void accept(Double i) {
|
|
if (Tripwire.ENABLED)
|
|
Tripwire.trip(getClass(), "{0} calling Sink.OfDouble.accept(Double)");
|
|
accept(i.doubleValue());
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Abstract {@code Sink} implementation for creating chains of
|
|
* sinks. The {@code begin}, {@code end}, and
|
|
* {@code cancellationRequested} methods are wired to chain to the
|
|
* downstream {@code Sink}. This implementation takes a downstream
|
|
* {@code Sink} of unknown input shape and produces a {@code Sink<T>}. The
|
|
* implementation of the {@code accept()} method must call the correct
|
|
* {@code accept()} method on the downstream {@code Sink}.
|
|
*/
|
|
abstract static class ChainedReference<T, E_OUT> implements Sink<T> {
|
|
protected final Sink<? super E_OUT> downstream;
|
|
|
|
public ChainedReference(Sink<? super E_OUT> downstream) {
|
|
this.downstream = Objects.requireNonNull(downstream);
|
|
}
|
|
|
|
@Override
|
|
public void begin(long size) {
|
|
downstream.begin(size);
|
|
}
|
|
|
|
@Override
|
|
public void end() {
|
|
downstream.end();
|
|
}
|
|
|
|
@Override
|
|
public boolean cancellationRequested() {
|
|
return downstream.cancellationRequested();
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Abstract {@code Sink} implementation designed for creating chains of
|
|
* sinks. The {@code begin}, {@code end}, and
|
|
* {@code cancellationRequested} methods are wired to chain to the
|
|
* downstream {@code Sink}. This implementation takes a downstream
|
|
* {@code Sink} of unknown input shape and produces a {@code Sink.OfInt}.
|
|
* The implementation of the {@code accept()} method must call the correct
|
|
* {@code accept()} method on the downstream {@code Sink}.
|
|
*/
|
|
abstract static class ChainedInt<E_OUT> implements Sink.OfInt {
|
|
protected final Sink<? super E_OUT> downstream;
|
|
|
|
public ChainedInt(Sink<? super E_OUT> downstream) {
|
|
this.downstream = Objects.requireNonNull(downstream);
|
|
}
|
|
|
|
@Override
|
|
public void begin(long size) {
|
|
downstream.begin(size);
|
|
}
|
|
|
|
@Override
|
|
public void end() {
|
|
downstream.end();
|
|
}
|
|
|
|
@Override
|
|
public boolean cancellationRequested() {
|
|
return downstream.cancellationRequested();
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Abstract {@code Sink} implementation designed for creating chains of
|
|
* sinks. The {@code begin}, {@code end}, and
|
|
* {@code cancellationRequested} methods are wired to chain to the
|
|
* downstream {@code Sink}. This implementation takes a downstream
|
|
* {@code Sink} of unknown input shape and produces a {@code Sink.OfLong}.
|
|
* The implementation of the {@code accept()} method must call the correct
|
|
* {@code accept()} method on the downstream {@code Sink}.
|
|
*/
|
|
abstract static class ChainedLong<E_OUT> implements Sink.OfLong {
|
|
protected final Sink<? super E_OUT> downstream;
|
|
|
|
public ChainedLong(Sink<? super E_OUT> downstream) {
|
|
this.downstream = Objects.requireNonNull(downstream);
|
|
}
|
|
|
|
@Override
|
|
public void begin(long size) {
|
|
downstream.begin(size);
|
|
}
|
|
|
|
@Override
|
|
public void end() {
|
|
downstream.end();
|
|
}
|
|
|
|
@Override
|
|
public boolean cancellationRequested() {
|
|
return downstream.cancellationRequested();
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Abstract {@code Sink} implementation designed for creating chains of
|
|
* sinks. The {@code begin}, {@code end}, and
|
|
* {@code cancellationRequested} methods are wired to chain to the
|
|
* downstream {@code Sink}. This implementation takes a downstream
|
|
* {@code Sink} of unknown input shape and produces a {@code Sink.OfDouble}.
|
|
* The implementation of the {@code accept()} method must call the correct
|
|
* {@code accept()} method on the downstream {@code Sink}.
|
|
*/
|
|
abstract static class ChainedDouble<E_OUT> implements Sink.OfDouble {
|
|
protected final Sink<? super E_OUT> downstream;
|
|
|
|
public ChainedDouble(Sink<? super E_OUT> downstream) {
|
|
this.downstream = Objects.requireNonNull(downstream);
|
|
}
|
|
|
|
@Override
|
|
public void begin(long size) {
|
|
downstream.begin(size);
|
|
}
|
|
|
|
@Override
|
|
public void end() {
|
|
downstream.end();
|
|
}
|
|
|
|
@Override
|
|
public boolean cancellationRequested() {
|
|
return downstream.cancellationRequested();
|
|
}
|
|
}
|
|
}
|