infernap12/jdk
1
0
Fork 0
mirror of https://github.com/openjdk/jdk.git synced 2026-02-26 10:10:19 +00:00
Code Issues Packages Projects Releases Wiki Activity

8015318: Extend Collector with 'finish' operation

Browse Source 
Reviewed-by: mduigou
This commit is contained in:
Brian Goetz 2013-06-28 16:26:54 -04:00 committed by Brian Goetz
parent 10f964a324
commit e11ec149ce
20 changed files with 909 additions and 544 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
jdk/src/share/classes/java/util/DoubleSummaryStatistics.java
View File

@ -25,6 +25,7 @@
package java.util;
import java.util.function.DoubleConsumer;
import java.util.stream.Collector;
/**
* A state object for collecting statistics such as count, min, max, sum, and
@ -35,24 +36,24 @@ import java.util.function.DoubleConsumer;
* summary statistics on a stream of doubles with:
* <pre> {@code
* DoubleSummaryStatistics stats = doubleStream.collect(DoubleSummaryStatistics::new,
* DoubleSummaryStatistics::accept,
* DoubleSummaryStatistics::combine);
* DoubleSummaryStatistics::accept,
* DoubleSummaryStatistics::combine);
* }</pre>
*
* <p>{@code DoubleSummaryStatistics} can be used as a
* {@linkplain java.util.stream.Stream#reduce(java.util.function.BinaryOperator) reduction}
* {@linkplain java.util.stream.Stream#collect(Collector) reduction}
* target for a {@linkplain java.util.stream.Stream stream}. For example:
*
* <pre> {@code
* DoubleSummaryStatistics stats = people.stream()
* .collect(Collectors.toDoubleSummaryStatistics(Person::getWeight));
* .collect(Collectors.summarizingDouble(Person::getWeight));
*}</pre>
*
* This computes, in a single pass, the count of people, as well as the minimum,
* maximum, sum, and average of their weights.
*
* @implNote This implementation is not thread safe. However, it is safe to use
* {@link java.util.stream.Collectors#toDoubleSummaryStatistics(java.util.function.ToDoubleFunction)
* {@link java.util.stream.Collectors#summarizingDouble(java.util.function.ToDoubleFunction)
* Collectors.toDoubleStatistics()} on a parallel stream, because the parallel
* implementation of {@link java.util.stream.Stream#collect Stream.collect()}
* provides the necessary partitioning, isolation, and merging of results for
@ -152,7 +153,7 @@ public class DoubleSummaryStatistics implements DoubleConsumer {
}
/**
* Returns the average of values recorded, or zero if no values have been
* Returns the arithmetic mean of values recorded, or zero if no values have been
* recorded. The average returned can vary depending upon the order in
* which values are recorded. This is due to accumulated rounding error in
* addition of values of differing magnitudes. Values sorted by increasing
@ -160,7 +161,7 @@ public class DoubleSummaryStatistics implements DoubleConsumer {
* value is a {@code NaN} or the sum is at any point a {@code NaN} then the
* average will be {@code NaN}.
*
* @return the average of values, or zero if none
* @return the arithmetic mean of values, or zero if none
*/
public final double getAverage() {
return getCount() > 0 ? getSum() / getCount() : 0.0d;

15
jdk/src/share/classes/java/util/IntSummaryStatistics.java
View File

@ -25,6 +25,7 @@
package java.util;
import java.util.function.IntConsumer;
import java.util.stream.Collector;
/**
* A state object for collecting statistics such as count, min, max, sum, and
@ -35,24 +36,24 @@ import java.util.function.IntConsumer;
* summary statistics on a stream of ints with:
* <pre> {@code
* IntSummaryStatistics stats = intStream.collect(IntSummaryStatistics::new,
* IntSummaryStatistics::accept,
* IntSummaryStatistics::combine);
* IntSummaryStatistics::accept,
* IntSummaryStatistics::combine);
* }</pre>
*
* <p>{@code IntSummaryStatistics} can be used as a
* {@linkplain java.util.stream.Stream#reduce(java.util.function.BinaryOperator) reduction}
* {@linkplain java.util.stream.Stream#collect(Collector) reduction}
* target for a {@linkplain java.util.stream.Stream stream}. For example:
*
* <pre> {@code
* IntSummaryStatistics stats = people.stream()
* .collect(Collectors.toIntSummaryStatistics(Person::getDependents));
* .collect(Collectors.summarizingInt(Person::getDependents));
*}</pre>
*
* This computes, in a single pass, the count of people, as well as the minimum,
* maximum, sum, and average of their number of dependents.
*
* @implNote This implementation is not thread safe. However, it is safe to use
* {@link java.util.stream.Collectors#toIntSummaryStatistics(java.util.function.ToIntFunction)
* {@link java.util.stream.Collectors#summarizingInt(java.util.function.ToIntFunction)
* Collectors.toIntStatistics()} on a parallel stream, because the parallel
* implementation of {@link java.util.stream.Stream#collect Stream.collect()}
* provides the necessary partitioning, isolation, and merging of results for
@ -140,10 +141,10 @@ public class IntSummaryStatistics implements IntConsumer {
}
/**
* Returns the average of values recorded, or zero if no values have been
* Returns the arithmetic mean of values recorded, or zero if no values have been
* recorded.
*
* @return the average of values, or zero if none
* @return the arithmetic mean of values, or zero if none
*/
public final double getAverage() {
return getCount() > 0 ? (double) getSum() / getCount() : 0.0d;

17
jdk/src/share/classes/java/util/LongSummaryStatistics.java
View File

@ -26,6 +26,7 @@ package java.util;
import java.util.function.IntConsumer;
import java.util.function.LongConsumer;
import java.util.stream.Collector;
/**
* A state object for collecting statistics such as count, min, max, sum, and
@ -36,24 +37,24 @@ import java.util.function.LongConsumer;
* summary statistics on a stream of longs with:
* <pre> {@code
* LongSummaryStatistics stats = longStream.collect(LongSummaryStatistics::new,
* LongSummaryStatistics::accept,
* LongSummaryStatistics::combine);
* LongSummaryStatistics::accept,
* LongSummaryStatistics::combine);
* }</pre>
*
* <p>{@code LongSummaryStatistics} can be used as a
* {@linkplain java.util.stream.Stream#reduce(java.util.function.BinaryOperator) reduction}
* {@linkplain java.util.stream.Stream#collect(Collector)} reduction}
* target for a {@linkplain java.util.stream.Stream stream}. For example:
*
* <pre> {@code
* LongSummaryStatistics stats = people.stream()
* .collect(Collectors.toLongSummaryStatistics(Person::getAge));
* .collect(Collectors.summarizingLong(Person::getAge));
*}</pre>
*
* This computes, in a single pass, the count of people, as well as the minimum,
* maximum, sum, and average of their ages in milliseconds.
* maximum, sum, and average of their ages.
*
* @implNote This implementation is not thread safe. However, it is safe to use
* {@link java.util.stream.Collectors#toLongSummaryStatistics(java.util.function.ToLongFunction)
* {@link java.util.stream.Collectors#summarizingLong(java.util.function.ToLongFunction)
* Collectors.toLongStatistics()} on a parallel stream, because the parallel
* implementation of {@link java.util.stream.Stream#collect Stream.collect()}
* provides the necessary partitioning, isolation, and merging of results for
@ -152,10 +153,10 @@ public class LongSummaryStatistics implements LongConsumer, IntConsumer {
}
/**
* Returns the average of values recorded, or zero if no values have been
* Returns the arithmetic mean of values recorded, or zero if no values have been
* recorded.
*
* @return The average of values, or zero if none
* @return The arithmetic mean of values, or zero if none
*/
public final double getAverage() {
return getCount() > 0 ? (double) getSum() / getCount() : 0.0d;

7
jdk/src/share/classes/java/util/StringJoiner.java
View File

@ -49,16 +49,17 @@ package java.util;
* <p>
* A {@code StringJoiner} may be employed to create formatted output from a
* {@link java.util.stream.Stream} using
* {@link java.util.stream.Collectors#toStringJoiner}. For example:
* {@link java.util.stream.Collectors#joining(CharSequence)}. For example:
*
* <pre> {@code
* List<Integer> numbers = Arrays.asList(1, 2, 3, 4);
* String commaSeparatedNumbers = numbers.stream()
* .map(i -> i.toString())
* .collect(Collectors.toStringJoiner(", ")).toString();
* .collect(Collectors.joining(", "));
* }</pre>
*
* @see java.util.stream.Collectors#toStringJoiner
* @see java.util.stream.Collectors#joining(CharSequence)
* @see java.util.stream.Collectors#joining(CharSequence, CharSequence, CharSequence)
* @since 1.8
*/
public final class StringJoiner {

222
jdk/src/share/classes/java/util/stream/Collector.java
View File

@ -25,40 +25,45 @@
package java.util.stream;
import java.util.Collections;
import java.util.EnumSet;
import java.util.Set;
import java.util.function.BiFunction;
import java.util.function.BiConsumer;
import java.util.function.BinaryOperator;
import java.util.function.Function;
import java.util.function.Supplier;
/**
* A <a href="package-summary.html#Reduction">reduction operation</a> that
* supports folding input elements into a cumulative result. The result may be
* a value or may be a mutable result container. Examples of operations
* accumulating results into a mutable result container include: accumulating
* input elements into a {@code Collection}; concatenating strings into a
* {@code StringBuilder}; computing summary information about elements such as
* sum, min, max, or average; computing "pivot table" summaries such as "maximum
* valued transaction by seller", etc. Reduction operations can be performed
* either sequentially or in parallel.
* folds input elements into a mutable result container, optionally transforming
* the accumulated result into a final representation after all input elements
* have been processed.
*
* <p>Examples of mutable reduction operations include:
* accumulating elements into a {@code Collection}; concatenating
* strings using a {@code StringBuilder}; computing summary information about
* elements such as sum, min, max, or average; computing "pivot table" summaries
* such as "maximum valued transaction by seller", etc. Reduction operations
* can be performed either sequentially or in parallel.
*
* <p>The following are examples of using the predefined {@code Collector}
* implementations in {@link Collectors} with the {@code Stream} API to perform
* mutable reduction tasks:
* <pre>{@code
* // Accumulate elements into a List
* List<String> list = stream.collect(Collectors.toList());
* // Accumulate names into a List
* List<String> list = people.stream().map(Person::getName).collect(Collectors.toList());
*
* // Accumulate elements into a TreeSet
* Set<String> list = stream.collect(Collectors.toCollection(TreeSet::new));
* // Accumulate names into a TreeSet
* Set<String> list = people.stream().map(Person::getName).collect(Collectors.toCollection(TreeSet::new));
*
* // Convert elements to strings and concatenate them, separated by commas
* String joined = stream.map(Object::toString)
* .collect(Collectors.toStringJoiner(", "))
* .toString();
* String joined = things.stream()
* .map(Object::toString)
* .collect(Collectors.joining(", "));
*
* // Find highest-paid employee
* Employee highestPaid = employees.stream()
* .collect(Collectors.maxBy(Comparators.comparing(Employee::getSalary)));
* .collect(Collectors.maxBy(Comparators.comparing(Employee::getSalary)))
* .get();
*
* // Group employees by department
* Map<Department, List<Employee>> byDept
@ -66,7 +71,7 @@ import java.util.function.Supplier;
* .collect(Collectors.groupingBy(Employee::getDepartment));
*
* // Find highest-paid employee by department
* Map<Department, Employee> highestPaidByDept
* Map<Department, Optional<Employee>> highestPaidByDept
* = employees.stream()
* .collect(Collectors.groupingBy(Employee::getDepartment,
* Collectors.maxBy(Comparators.comparing(Employee::getSalary))));
@ -74,43 +79,42 @@ import java.util.function.Supplier;
* // Partition students into passing and failing
* Map<Boolean, List<Student>> passingFailing =
* students.stream()
* .collect(Collectors.partitioningBy(s -> s.getGrade() >= PASS_THRESHOLD);
* .collect(Collectors.partitioningBy(s -> s.getGrade() >= PASS_THRESHOLD));
*
* }</pre>
*
* <p>A {@code Collector} is specified by three functions that work together to
* manage a result or result container. They are: creation of an initial
* result, incorporating a new data element into a result, and combining two
* results into one. The last function -- combining two results into one -- is
* used during parallel operations, where subsets of the input are accumulated
* in parallel, and then the subresults merged into a combined result. The
* result may be a mutable container or a value. If the result is mutable, the
* accumulation and combination functions may either mutate their left argument
* and return that (such as adding elements to a collection), or return a new
* result, in which case it should not perform any mutation.
* <p>A {@code Collector} is specified by four functions that work together to
* accumulate entries into a mutable result container, and optionally perform
* a final transform on the result. They are: creation of a new result container,
* incorporating a new data element into a result container, combining two
* result containers into one, and performing a final transform on the container.
* The combiner function is used during parallel operations, where
* subsets of the input are accumulated into separate result
* containers, and then the subresults merged into a combined result. The
* combiner function may merge one set of subresults into the other and return
* that, or it may return a new object to describe the combined results.
*
* <p>Collectors also have a set of characteristics, including
* {@link Characteristics#CONCURRENT} and
* {@link Characteristics#STRICTLY_MUTATIVE}. These characteristics provide
* <p>Collectors also have a set of characteristics, such as
* {@link Characteristics#CONCURRENT}. These characteristics provide
* hints that can be used by a reduction implementation to provide better
* performance.
*
* <p>Libraries that implement reduction based on {@code Collector}, such as
* {@link Stream#collect(Collector)}, must adhere to the following constraints:
* <ul>
* <li>The first argument passed to the accumulator function, and both
* arguments passed to the combiner function, must be the result of a
* previous invocation of {@link #resultSupplier()}, {@link #accumulator()},
* or {@link #combiner()}.</li>
* <li>The first argument passed to the accumulator function, both
* arguments passed to the combiner function, and the argument passed to the
* finisher function must be the result of a previous invocation of the
* result supplier, accumulator, or combiner functions.</li>
* <li>The implementation should not do anything with the result of any of
* the result supplier, accumulator, or combiner functions other than to
* pass them again to the accumulator or combiner functions, or return them
* to the caller of the reduction operation.</li>
* <li>If a result is passed to the accumulator or combiner function, and
* the same object is not returned from that function, it is never used
* again.</li>
* <li>Once a result is passed to the combiner function, it is never passed
* to the accumulator function again.</li>
* pass them again to the accumulator, combiner, or finisher functions,
* or return them to the caller of the reduction operation.</li>
* <li>If a result is passed to the combiner or finisher
* function, and the same object is not returned from that function, it is
* never used again.</li>
* <li>Once a result is passed to the combiner or finisher function, it
* is never passed to the accumulator function again.</li>
* <li>For non-concurrent collectors, any result returned from the result
* supplier, accumulator, or combiner functions must be serially
* thread-confined. This enables collection to occur in parallel without
@ -132,11 +136,10 @@ import java.util.function.Supplier;
* Performing a reduction operation with a {@code Collector} should produce a
* result equivalent to:
* <pre>{@code
* BiFunction<R,T,R> accumulator = collector.accumulator();
* R result = collector.resultSupplier().get();
* R container = collector.supplier().get();
* for (T t : data)
* result = accumulator.apply(result, t);
* return result;
* collector.accumulator().accept(container, t);
* return collector.finisher().apply(container);
* }</pre>
*
* <p>However, the library is free to partition the input, perform the reduction
@ -149,7 +152,7 @@ import java.util.function.Supplier;
* is accumulating elements into a {@code TreeSet}. In this case, the {@code
* resultSupplier()} function is {@code () -> new Treeset<T>()}, the
* {@code accumulator} function is
* {@code (set, element) -> { set.add(element); return set; }}, and the combiner
* {@code (set, element) -> set.add(element) }, and the combiner
* function is {@code (left, right) -> { left.addAll(right); return left; }}.
* (This behavior is implemented by
* {@code Collectors.toCollection(TreeSet::new)}).
@ -159,51 +162,49 @@ import java.util.function.Supplier;
* @see Stream#collect(Collector)
* @see Collectors
*
* @param <T> the type of input element to the collect operation
* @param <R> the result type of the collect operation
* @param <T> the type of input elements to the reduction operation
* @param <A> the mutable accumulation type of the reduction operation (often
* hidden as an implementation detail)
* @param <R> the result type of the reduction operation
* @since 1.8
*/
public interface Collector<T, R> {
public interface Collector<T, A, R> {
/**
* A function that creates and returns a new result that represents
* "no values". If the accumulator or combiner functions may mutate their
* arguments, this must be a new, empty result container.
* A function that creates and returns a new mutable result container.
*
* @return a function which, when invoked, returns a result representing
* "no values"
* @return a function which returns a new, mutable result container
*/
Supplier<R> resultSupplier();
Supplier<A> supplier();
/**
* A function that folds a new value into a cumulative result. The result
* may be a mutable result container or a value. The accumulator function
* may modify a mutable container and return it, or create a new result and
* return that, but if it returns a new result object, it must not modify
* any of its arguments.
* A function that folds a new value into a mutable result container.
*
* <p>If the collector has the {@link Characteristics#STRICTLY_MUTATIVE}
* characteristic, then the accumulator function <em>must</em> always return
* its first argument, after possibly mutating its state.
*
* @return a function which folds a new value into a cumulative result
* @return a function which folds a new value into a mutable result container
*/
BiFunction<R, T, R> accumulator();
BiConsumer<A, T> accumulator();
/**
* A function that accepts two partial results and merges them. The
* combiner function may fold state from one argument into the other and
* return that, or may return a new result object, but if it returns
* a new result object, it must not modify the state of either of its
* arguments.
*
* <p>If the collector has the {@link Characteristics#STRICTLY_MUTATIVE}
* characteristic, then the combiner function <em>must</em> always return
* its first argument, after possibly mutating its state.
* return that, or may return a new result object.
*
* @return a function which combines two partial results into a cumulative
* result
*/
BinaryOperator<R> combiner();
BinaryOperator<A> combiner();
/**
* Perform the final transformation from the intermediate accumulation type
* {@code A} to the final result representation {@code R}.
*
* <p>If the characteristic {@code IDENTITY_TRANSFORM} is
* set, this function may be presumed to be an identity transform with an
* unchecked cast from {@code A} to {@code R}.
*
* @return a function which transforms the intermediate result to the final
* result
*/
Function<A, R> finisher();
/**
* Returns a {@code Set} of {@code Collector.Characteristics} indicating
@ -213,6 +214,62 @@ public interface Collector<T, R> {
*/
Set<Characteristics> characteristics();
/**
* Returns a new {@code Collector} described by the given {@code supplier},
* {@code accumulator}, and {@code combiner} functions. The resulting
* {@code Collector} has the {@code Collector.Characteristics.IDENTITY_FINISH}
* characteristic.
*
* @param supplier The supplier function for the new collector
* @param accumulator The accumulator function for the new collector
* @param combiner The combiner function for the new collector
* @param characteristics The collector characteristics for the new
* collector
* @param <T> The type of input elements for the new collector
* @param <R> The type of intermediate accumulation result, and final result,
* for the new collector
* @return the new {@code Collector}
*/
public static<T, R> Collector<T, R, R> of(Supplier<R> supplier,
BiConsumer<R, T> accumulator,
BinaryOperator<R> combiner,
Characteristics... characteristics) {
Set<Characteristics> cs = (characteristics.length == 0)
? Collectors.CH_ID
: Collections.unmodifiableSet(EnumSet.of(Collector.Characteristics.IDENTITY_FINISH,
characteristics));
return new Collectors.CollectorImpl<>(supplier, accumulator, combiner, cs);
}
/**
* Returns a new {@code Collector} described by the given {@code supplier},
* {@code accumulator}, {@code combiner}, and {@code finisher} functions.
*
* @param supplier The supplier function for the new collector
* @param accumulator The accumulator function for the new collector
* @param combiner The combiner function for the new collector
* @param finisher The finisher function for the new collector
* @param characteristics The collector characteristics for the new
* collector
* @param <T> The type of input elements for the new collector
* @param <A> The intermediate accumulation type of the new collector
* @param <R> The final result type of the new collector
* @return the new {@code Collector}
*/
public static<T, A, R> Collector<T, A, R> of(Supplier<A> supplier,
BiConsumer<A, T> accumulator,
BinaryOperator<A> combiner,
Function<A, R> finisher,
Characteristics... characteristics) {
Set<Characteristics> cs = Collectors.CH_NOID;
if (characteristics.length > 0) {
cs = EnumSet.noneOf(Characteristics.class);
Collections.addAll(cs, characteristics);
cs = Collections.unmodifiableSet(cs);
}
return new Collectors.CollectorImpl<>(supplier, accumulator, combiner, finisher, cs);
}
/**
* Characteristics indicating properties of a {@code Collector}, which can
* be used to optimize reduction implementations.
@ -222,8 +279,7 @@ public interface Collector<T, R> {
* Indicates that this collector is <em>concurrent</em>, meaning that
* the result container can support the accumulator function being
* called concurrently with the same result container from multiple
* threads. Concurrent collectors must also always have the
* {@code STRICTLY_MUTATIVE} characteristic.
* threads.
*
* <p>If a {@code CONCURRENT} collector is not also {@code UNORDERED},
* then it should only be evaluated concurrently if applied to an
@ -238,12 +294,10 @@ public interface Collector<T, R> {
UNORDERED,
/**
* Indicates that this collector operates by strict mutation of its
* result container. This means that the {@link #accumulator()} and
* {@link #combiner()} functions will always modify the state of and
* return their first argument, rather than returning a different result
* container.
* Indicates that the finisher function is the identity function and
* can be elided. If set, it must be the case that an unchecked cast
* from A to R will succeed.
*/
STRICTLY_MUTATIVE
IDENTITY_FINISH
}
}

865
jdk/src/share/classes/java/util/stream/Collectors.java
View File

File diff suppressed because it is too large Load Diff

2
jdk/src/share/classes/java/util/stream/DelegatingStream.java
View File

@ -209,7 +209,7 @@ public class DelegatingStream<T> implements Stream<T> {
}
@Override
public <R> R collect(Collector<? super T, R> collector) {
public <R, A> R collect(Collector<? super T, A, ? extends R> collector) {
return delegate.collect(collector);
}

2
jdk/src/share/classes/java/util/stream/DoubleStream.java
View File

@ -527,7 +527,7 @@ public interface DoubleStream extends BaseStream<Double, DoubleStream> {
long count();
/**
* Returns an {@code OptionalDouble} describing the average of elements of
* Returns an {@code OptionalDouble} describing the arithmetic mean of elements of
* this stream, or an empty optional if this stream is empty. The average
* returned can vary depending upon the order in which elements are
* encountered. This is due to accumulated rounding error in addition of

2
jdk/src/share/classes/java/util/stream/IntStream.java
View File

@ -517,7 +517,7 @@ public interface IntStream extends BaseStream<Integer, IntStream> {
long count();
/**
* Returns an {@code OptionalDouble} describing the average of elements of
* Returns an {@code OptionalDouble} describing the arithmetic mean of elements of
* this stream, or an empty optional if this stream is empty. This is a
* special case of a
* <a href="package-summary.html#MutableReduction">reduction</a>.

2
jdk/src/share/classes/java/util/stream/LongStream.java
View File

@ -517,7 +517,7 @@ public interface LongStream extends BaseStream<Long, LongStream> {
long count();
/**
* Returns an {@code OptionalDouble} describing the average of elements of
* Returns an {@code OptionalDouble} describing the arithmetic mean of elements of
* this stream, or an empty optional if this stream is empty. This is a
* special case of a
* <a href="package-summary.html#MutableReduction">reduction</a>.

22
jdk/src/share/classes/java/util/stream/ReduceOps.java
View File

@ -148,17 +148,17 @@ final class ReduceOps {
* reference values.
*
* @param <T> the type of the input elements
* @param <R> the type of the result
* @param <I> the type of the intermediate reduction result
* @param collector a {@code Collector} defining the reduction
* @return a {@code ReduceOp} implementing the reduction
*/
public static <T,R> TerminalOp<T, R>
makeRef(Collector<? super T,R> collector) {
Supplier<R> supplier = Objects.requireNonNull(collector).resultSupplier();
BiFunction<R, ? super T, R> accumulator = collector.accumulator();
BinaryOperator<R> combiner = collector.combiner();
class ReducingSink extends Box<R>
implements AccumulatingSink<T, R, ReducingSink> {
public static <T, I> TerminalOp<T, I>
makeRef(Collector<? super T, I, ?> collector) {
Supplier<I> supplier = Objects.requireNonNull(collector).supplier();
BiConsumer<I, ? super T> accumulator = collector.accumulator();
BinaryOperator<I> combiner = collector.combiner();
class ReducingSink extends Box<I>
implements AccumulatingSink<T, I, ReducingSink> {
@Override
public void begin(long size) {
state = supplier.get();
@ -166,9 +166,7 @@ final class ReduceOps {
@Override
public void accept(T t) {
R newResult = accumulator.apply(state, t);
if (state != newResult)
state = newResult;
accumulator.accept(state, t);
}
@Override
@ -176,7 +174,7 @@ final class ReduceOps {
state = combiner.apply(state, other.state);
}
}
return new ReduceOp<T, R, ReducingSink>(StreamShape.REFERENCE) {
return new ReduceOp<T, I, ReducingSink>(StreamShape.REFERENCE) {
@Override
public ReducingSink makeSink() {
return new ReducingSink();

17
jdk/src/share/classes/java/util/stream/ReferencePipeline.java
View File

@ -490,16 +490,21 @@ abstract class ReferencePipeline<P_IN, P_OUT>
}
@Override
public final <R> R collect(Collector<? super P_OUT, R> collector) {
public final <R, A> R collect(Collector<? super P_OUT, A, ? extends R> collector) {
A container;
if (isParallel()
&& (collector.characteristics().contains(Collector.Characteristics.CONCURRENT))
&& (!isOrdered() || collector.characteristics().contains(Collector.Characteristics.UNORDERED))) {
R container = collector.resultSupplier().get();
BiFunction<R, ? super P_OUT, R> accumulator = collector.accumulator();
forEach(u -> accumulator.apply(container, u));
return container;
container = collector.supplier().get();
BiConsumer<A, ? super P_OUT> accumulator = collector.accumulator();
forEach(u -> accumulator.accept(container, u));
}
return evaluate(ReduceOps.makeRef(collector));
else {
container = evaluate(ReduceOps.makeRef(collector));
}
return collector.characteristics().contains(Collector.Characteristics.IDENTITY_FINISH)
? (R) container
: collector.finisher().apply(container);
}
@Override

3
jdk/src/share/classes/java/util/stream/Stream.java
View File

@ -651,12 +651,13 @@ public interface Stream<T> extends BaseStream<T, Stream<T>> {
* }</pre>
*
* @param <R> the type of the result
* @param <A> the intermediate accumulation type of the {@code Collector}
* @param collector the {@code Collector} describing the reduction
* @return the result of the reduction
* @see #collect(Supplier, BiConsumer, BiConsumer)
* @see Collectors
*/
<R> R collect(Collector<? super T, R> collector);
<R, A> R collect(Collector<? super T, A, ? extends R> collector);
/**
* Returns the minimum element of this stream according to the provided

4
jdk/src/share/classes/java/util/stream/package-info.java
View File

@ -547,7 +547,7 @@
* List<String> l = new ArrayList(Arrays.asList("one", "two"));
* Stream<String> sl = l.stream();
* l.add("three");
* String s = sl.collect(toStringJoiner(" ")).toString();
* String s = sl.collect(joining(" "));
* }</pre>
* First a list is created consisting of two strings: "one"; and "two". Then a stream is created from that list.
* Next the list is modified by adding a third string: "three". Finally the elements of the stream are collected
@ -557,7 +557,7 @@
* <pre>{@code
* List<String> l = new ArrayList(Arrays.asList("one", "two"));
* Stream<String> sl = l.stream();
* String s = sl.peek(s -> l.add("BAD LAMBDA")).collect(toStringJoiner(" ")).toString();
* String s = sl.peek(s -> l.add("BAD LAMBDA")).collect(joining(" "));
* }</pre>
* then a {@code ConcurrentModificationException} will be thrown since the {@code peek} operation will attempt
* to add the string "BAD LAMBDA" to the list after the terminal operation has commenced.

6
jdk/test/java/util/stream/test/org/openjdk/tests/java/util/FillableStringTest.java
View File

@ -40,17 +40,17 @@ public class FillableStringTest {
}
public void testStringBuilder() {
String s = generate().collect(Collectors.toStringBuilder()).toString();
String s = generate().collect(Collectors.joining());
assertEquals(s, "THREEFOURFIVE");
}
public void testStringBuffer() {
String s = generate().collect(Collectors.toStringBuilder()).toString();
String s = generate().collect(Collectors.joining());
assertEquals(s, "THREEFOURFIVE");
}
public void testStringJoiner() {
String s = generate().collect(Collectors.toStringJoiner("-")).toString();
String s = generate().collect(Collectors.joining("-"));
assertEquals(s, "THREE-FOUR-FIVE");
}
}

12
jdk/test/java/util/stream/test/org/openjdk/tests/java/util/stream/GroupByOpTest.java
View File

@ -36,7 +36,6 @@ import java.util.stream.Collectors;
import java.util.stream.LambdaTestHelpers;
import java.util.stream.OpTestCase;
import java.util.stream.Stream;
import java.util.stream.StreamOpFlagTestHelper;
import java.util.stream.StreamTestDataProvider;
import java.util.stream.TestData;
@ -59,13 +58,14 @@ import static java.util.stream.LambdaTestHelpers.pTrue;
public class GroupByOpTest extends OpTestCase {
public void testBypassCollect() {
Collector<Integer, Map<Boolean, List<Integer>>> collector
= Collectors.groupingBy(LambdaTestHelpers.forPredicate(pEven, true, false));
@SuppressWarnings("unchecked")
Collector<Integer, Map<Boolean, List<Integer>>, Map<Boolean, List<Integer>>> collector
= (Collector<Integer, Map<Boolean, List<Integer>>, Map<Boolean, List<Integer>>>) Collectors.groupingBy(LambdaTestHelpers.forPredicate(pEven, true, false));
Map<Boolean, List<Integer>> m = collector.resultSupplier().get();
Map<Boolean, List<Integer>> m = collector.supplier().get();
int[] ints = countTo(10).stream().mapToInt(e -> (int) e).toArray();
for (int i : ints)
m = collector.accumulator().apply(m, i);
collector.accumulator().accept(m, i);
assertEquals(2, m.keySet().size());
for(Collection<Integer> group : m.values()) {
@ -130,7 +130,7 @@ public class GroupByOpTest extends OpTestCase {
// - Total number of values equals size of data
for (MapperData<Integer, ?> md : getMapperData(data)) {
Collector<Integer, Map<Object, List<Integer>>> tab = Collectors.groupingBy(md.m);
Collector<Integer, ?, Map<Object, List<Integer>>> tab = Collectors.groupingBy(md.m);
Map<Object, List<Integer>> result =
withData(data)
.terminal(s -> s, s -> s.collect(tab))

12
jdk/test/java/util/stream/test/org/openjdk/tests/java/util/stream/SummaryStatisticsTest.java
View File

@ -43,9 +43,9 @@ import static java.util.stream.LambdaTestHelpers.countTo;
public class SummaryStatisticsTest extends OpTestCase {
public void testIntStatistics() {
List<IntSummaryStatistics> instances = new ArrayList<>();
instances.add(countTo(1000).stream().collect(Collectors.toIntSummaryStatistics(i -> i)));
instances.add(countTo(1000).stream().collect(Collectors.summarizingInt(i -> i)));
instances.add(countTo(1000).stream().mapToInt(i -> i).summaryStatistics());
instances.add(countTo(1000).parallelStream().collect(Collectors.toIntSummaryStatistics(i -> i)));
instances.add(countTo(1000).parallelStream().collect(Collectors.summarizingInt(i -> i)));
instances.add(countTo(1000).parallelStream().mapToInt(i -> i).summaryStatistics());
for (IntSummaryStatistics stats : instances) {
@ -58,9 +58,9 @@ public class SummaryStatisticsTest extends OpTestCase {
public void testLongStatistics() {
List<LongSummaryStatistics> instances = new ArrayList<>();
instances.add(countTo(1000).stream().collect(Collectors.toLongSummaryStatistics(i -> i)));
instances.add(countTo(1000).stream().collect(Collectors.summarizingLong(i -> i)));
instances.add(countTo(1000).stream().mapToLong(i -> i).summaryStatistics());
instances.add(countTo(1000).parallelStream().collect(Collectors.toLongSummaryStatistics(i -> i)));
instances.add(countTo(1000).parallelStream().collect(Collectors.summarizingLong(i -> i)));
instances.add(countTo(1000).parallelStream().mapToLong(i -> i).summaryStatistics());
for (LongSummaryStatistics stats : instances) {
@ -73,9 +73,9 @@ public class SummaryStatisticsTest extends OpTestCase {
public void testDoubleStatistics() {
List<DoubleSummaryStatistics> instances = new ArrayList<>();
instances.add(countTo(1000).stream().collect(Collectors.toDoubleSummaryStatistics(i -> i)));
instances.add(countTo(1000).stream().collect(Collectors.summarizingDouble(i -> i)));
instances.add(countTo(1000).stream().mapToDouble(i -> i).summaryStatistics());
instances.add(countTo(1000).parallelStream().collect(Collectors.toDoubleSummaryStatistics(i -> i)));
instances.add(countTo(1000).parallelStream().collect(Collectors.summarizingDouble(i -> i)));
instances.add(countTo(1000).parallelStream().mapToDouble(i -> i).summaryStatistics());
for (DoubleSummaryStatistics stats : instances) {

222
jdk/test/java/util/stream/test/org/openjdk/tests/java/util/stream/TabulatorsTest.java
View File

@ -23,13 +23,17 @@
package org.openjdk.tests.java.util.stream;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.StringJoiner;
import java.util.TreeMap;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentSkipListMap;
@ -53,7 +57,10 @@ import static java.util.stream.Collectors.groupingByConcurrent;
import static java.util.stream.Collectors.partitioningBy;
import static java.util.stream.Collectors.reducing;
import static java.util.stream.Collectors.toCollection;
import static java.util.stream.Collectors.toConcurrentMap;
import static java.util.stream.Collectors.toList;
import static java.util.stream.Collectors.toMap;
import static java.util.stream.Collectors.toSet;
import static java.util.stream.LambdaTestHelpers.assertContents;
import static java.util.stream.LambdaTestHelpers.assertContentsUnordered;
import static java.util.stream.LambdaTestHelpers.mDoubler;
@ -65,16 +72,6 @@ import static java.util.stream.LambdaTestHelpers.mDoubler;
*/
@SuppressWarnings({"rawtypes", "unchecked"})
public class TabulatorsTest extends OpTestCase {
// There are 8 versions of groupingBy:
// groupingBy: { map supplier, not } x { downstream collector, not } x { concurrent, not }
// There are 2 versions of partition: { map supplier, not }
// There are 4 versions of toMap
// mappedTo(function, mapSupplier?, mergeFunction?)
// Each variety needs at least one test
// Plus a variety of multi-level tests (groupBy(..., partition), partition(..., groupBy))
// Plus negative tests for mapping to null
// Each test should be matched by a nest of asserters (see TabulationAssertion...)
private static abstract class TabulationAssertion<T, U> {
abstract void assertValue(U value,
@ -101,7 +98,7 @@ public class TabulatorsTest extends OpTestCase {
boolean ordered) throws ReflectiveOperationException {
if (!clazz.isAssignableFrom(map.getClass()))
fail(String.format("Class mismatch in GroupedMapAssertion: %s, %s", clazz, map.getClass()));
assertContentsUnordered(map.keySet(), source.get().map(classifier).collect(Collectors.toSet()));
assertContentsUnordered(map.keySet(), source.get().map(classifier).collect(toSet()));
for (Map.Entry<K, ? extends V> entry : map.entrySet()) {
K key = entry.getKey();
downstream.assertValue(entry.getValue(),
@ -111,6 +108,39 @@ public class TabulatorsTest extends OpTestCase {
}
}
static class ToMapAssertion<T, K, V, M extends Map<K,V>> extends TabulationAssertion<T, M> {
private final Class<? extends Map> clazz;
private final Function<T, K> keyFn;
private final Function<T, V> valueFn;
private final BinaryOperator<V> mergeFn;
ToMapAssertion(Function<T, K> keyFn,
Function<T, V> valueFn,
BinaryOperator<V> mergeFn,
Class<? extends Map> clazz) {
this.clazz = clazz;
this.keyFn = keyFn;
this.valueFn = valueFn;
this.mergeFn = mergeFn;
}
@Override
void assertValue(M map, Supplier<Stream<T>> source, boolean ordered) throws ReflectiveOperationException {
Set<K> uniqueKeys = source.get().map(keyFn).collect(toSet());
assertTrue(clazz.isAssignableFrom(map.getClass()));
assertEquals(uniqueKeys, map.keySet());
source.get().forEach(t -> {
K key = keyFn.apply(t);
V v = source.get()
.filter(e -> key.equals(keyFn.apply(e)))
.map(valueFn)
.reduce(mergeFn)
.get();
assertEquals(map.get(key), v);
});
}
}
static class PartitionAssertion<T, D> extends TabulationAssertion<T, Map<Boolean,D>> {
private final Predicate<T> predicate;
private final TabulationAssertion<T,D> downstream;
@ -204,7 +234,7 @@ public class TabulatorsTest extends OpTestCase {
private <T> ResultAsserter<T> mapTabulationAsserter(boolean ordered) {
return (act, exp, ord, par) -> {
if (par & (!ordered || !ord)) {
if (par && (!ordered || !ord)) {
TabulatorsTest.nestedMapEqualityAssertion(act, exp);
}
else {
@ -215,7 +245,7 @@ public class TabulatorsTest extends OpTestCase {
private<T, M extends Map>
void exerciseMapTabulation(TestData<T, Stream<T>> data,
Collector<T, ? extends M> collector,
Collector<T, ?, ? extends M> collector,
TabulationAssertion<T, M> assertion)
throws ReflectiveOperationException {
boolean ordered = !collector.characteristics().contains(Collector.Characteristics.UNORDERED);
@ -248,6 +278,172 @@ public class TabulatorsTest extends OpTestCase {
assertEquals(o1, o2);
}
private<T, R> void assertCollect(TestData.OfRef<T> data,
Collector<T, ?, R> collector,
Function<Stream<T>, R> streamReduction) {
R check = streamReduction.apply(data.stream());
withData(data).terminal(s -> s.collect(collector)).expectedResult(check).exercise();
}
@Test(dataProvider = "StreamTestData<Integer>", dataProviderClass = StreamTestDataProvider.class)
public void testReduce(String name, TestData.OfRef<Integer> data) throws ReflectiveOperationException {
assertCollect(data, Collectors.reducing(0, Integer::sum),
s -> s.reduce(0, Integer::sum));
assertCollect(data, Collectors.reducing(Integer.MAX_VALUE, Integer::min),
s -> s.min(Integer::compare).orElse(Integer.MAX_VALUE));
assertCollect(data, Collectors.reducing(Integer.MIN_VALUE, Integer::max),
s -> s.max(Integer::compare).orElse(Integer.MIN_VALUE));
assertCollect(data, Collectors.reducing(Integer::sum),
s -> s.reduce(Integer::sum));
assertCollect(data, Collectors.minBy(Comparator.naturalOrder()),
s -> s.min(Integer::compare));
assertCollect(data, Collectors.maxBy(Comparator.naturalOrder()),
s -> s.max(Integer::compare));
assertCollect(data, Collectors.reducing(0, x -> x*2, Integer::sum),
s -> s.map(x -> x*2).reduce(0, Integer::sum));
assertCollect(data, Collectors.summingLong(x -> x * 2L),
s -> s.map(x -> x*2L).reduce(0L, Long::sum));
assertCollect(data, Collectors.summingInt(x -> x * 2),
s -> s.map(x -> x*2).reduce(0, Integer::sum));
assertCollect(data, Collectors.summingDouble(x -> x * 2.0d),
s -> s.map(x -> x * 2.0d).reduce(0.0d, Double::sum));
assertCollect(data, Collectors.averagingInt(x -> x * 2),
s -> s.mapToInt(x -> x * 2).average().orElse(0));
assertCollect(data, Collectors.averagingLong(x -> x * 2),
s -> s.mapToLong(x -> x * 2).average().orElse(0));
assertCollect(data, Collectors.averagingDouble(x -> x * 2),
s -> s.mapToDouble(x -> x * 2).average().orElse(0));
// Test explicit Collector.of
Collector<Integer, long[], Double> avg2xint = Collector.of(() -> new long[2],
(a, b) -> {
a[0] += b * 2;
a[1]++;
},
(a, b) -> {
a[0] += b[0];
a[1] += b[1];
return a;
},
a -> a[1] == 0 ? 0.0d : (double) a[0] / a[1]);
assertCollect(data, avg2xint,
s -> s.mapToInt(x -> x * 2).average().orElse(0));
}
@Test(dataProvider = "StreamTestData<Integer>", dataProviderClass = StreamTestDataProvider.class)
public void testJoin(String name, TestData.OfRef<Integer> data) throws ReflectiveOperationException {
withData(data)
.terminal(s -> s.map(Object::toString).collect(Collectors.joining()))
.expectedResult(join(data, ""))
.exercise();
Collector<String, StringBuilder, String> likeJoining = Collector.of(StringBuilder::new, StringBuilder::append, (sb1, sb2) -> sb1.append(sb2.toString()), StringBuilder::toString);
withData(data)
.terminal(s -> s.map(Object::toString).collect(likeJoining))
.expectedResult(join(data, ""))
.exercise();
withData(data)
.terminal(s -> s.map(Object::toString).collect(Collectors.joining(",")))
.expectedResult(join(data, ","))
.exercise();
withData(data)
.terminal(s -> s.map(Object::toString).collect(Collectors.joining(",", "[", "]")))
.expectedResult("[" + join(data, ",") + "]")
.exercise();
withData(data)
.terminal(s -> s.map(Object::toString)
.collect(StringBuilder::new, StringBuilder::append, StringBuilder::append)
.toString())
.expectedResult(join(data, ""))
.exercise();
withData(data)
.terminal(s -> s.map(Object::toString)
.collect(() -> new StringJoiner(","),
(sj, cs) -> sj.add(cs),
(j1, j2) -> j1.merge(j2))
.toString())
.expectedResult(join(data, ","))
.exercise();
withData(data)
.terminal(s -> s.map(Object::toString)
.collect(() -> new StringJoiner(",", "[", "]"),
(sj, cs) -> sj.add(cs),
(j1, j2) -> j1.merge(j2))
.toString())
.expectedResult("[" + join(data, ",") + "]")
.exercise();
}
private<T> String join(TestData.OfRef<T> data, String delim) {
StringBuilder sb = new StringBuilder();
boolean first = true;
for (T i : data) {
if (!first)
sb.append(delim);
sb.append(i.toString());
first = false;
}
return sb.toString();
}
@Test(dataProvider = "StreamTestData<Integer>", dataProviderClass = StreamTestDataProvider.class)
public void testSimpleToMap(String name, TestData.OfRef<Integer> data) throws ReflectiveOperationException {
Function<Integer, Integer> keyFn = i -> i * 2;
Function<Integer, Integer> valueFn = i -> i * 4;
List<Integer> dataAsList = Arrays.asList(data.stream().toArray(Integer[]::new));
Set<Integer> dataAsSet = new HashSet<>(dataAsList);
BinaryOperator<Integer> sum = Integer::sum;
for (BinaryOperator<Integer> op : Arrays.asList((u, v) -> u,
(u, v) -> v,
sum)) {
try {
exerciseMapTabulation(data, toMap(keyFn, valueFn),
new ToMapAssertion<>(keyFn, valueFn, op, HashMap.class));
if (dataAsList.size() != dataAsSet.size())
fail("Expected ISE on input with duplicates");
}
catch (IllegalStateException e) {
if (dataAsList.size() == dataAsSet.size())
fail("Expected no ISE on input without duplicates");
}
exerciseMapTabulation(data, toMap(keyFn, valueFn, op),
new ToMapAssertion<>(keyFn, valueFn, op, HashMap.class));
exerciseMapTabulation(data, toMap(keyFn, valueFn, op, TreeMap::new),
new ToMapAssertion<>(keyFn, valueFn, op, TreeMap.class));
}
// For concurrent maps, only use commutative merge functions
try {
exerciseMapTabulation(data, toConcurrentMap(keyFn, valueFn),
new ToMapAssertion<>(keyFn, valueFn, sum, ConcurrentHashMap.class));
if (dataAsList.size() != dataAsSet.size())
fail("Expected ISE on input with duplicates");
}
catch (IllegalStateException e) {
if (dataAsList.size() == dataAsSet.size())
fail("Expected no ISE on input without duplicates");
}
exerciseMapTabulation(data, toConcurrentMap(keyFn, valueFn, sum),
new ToMapAssertion<>(keyFn, valueFn, sum, ConcurrentHashMap.class));
exerciseMapTabulation(data, toConcurrentMap(keyFn, valueFn, sum, ConcurrentSkipListMap::new),
new ToMapAssertion<>(keyFn, valueFn, sum, ConcurrentSkipListMap.class));
}
@Test(dataProvider = "StreamTestData<Integer>", dataProviderClass = StreamTestDataProvider.class)
public void testSimpleGroupBy(String name, TestData.OfRef<Integer> data) throws ReflectiveOperationException {
Function<Integer, Integer> classifier = i -> i % 3;

2
jdk/test/jdk/lambda/MethodReferenceTestInstanceMethod.java
View File

@ -47,7 +47,7 @@ public class MethodReferenceTestInstanceMethod {
}
public void testStringBuffer() {
String s = generate().collect(Collectors.toStringBuilder()).toString();
String s = generate().collect(Collectors.joining());
assertEquals(s, "THREEFOURFIVE");
}

4
jdk/test/jdk/lambda/separate/TestHarness.java
View File

@ -119,7 +119,7 @@ public class TestHarness {
Class stub = new Class(specimen.getName(), cm);
String params =
Arrays.asList(args).stream().collect(Collectors.toStringJoiner(", ")).toString();
Arrays.asList(args).stream().collect(Collectors.joining(", ")).toString();
ConcreteMethod sm = new ConcreteMethod(
method.getReturnType(), method.getName(),
@ -150,7 +150,7 @@ public class TestHarness {
null, Arrays.asList((Method)method));
Class cstub = new Class(specimen.getName());
String params = Arrays.asList(args).stream().collect(Collectors.toStringJoiner(", ")).toString();
String params = Arrays.asList(args).stream().collect(Collectors.joining(", ")).toString();
ConcreteMethod sm = new ConcreteMethod(
"int", SourceModel.stdMethodName,