/* * 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.Arrays; import java.util.Comparator; import java.util.Iterator; import java.util.Objects; import java.util.Optional; import java.util.Spliterator; import java.util.Spliterators; import java.util.function.BiConsumer; import java.util.function.BiFunction; import java.util.function.BinaryOperator; import java.util.function.Consumer; import java.util.function.Function; import java.util.function.IntFunction; import java.util.function.Predicate; import java.util.function.Supplier; import java.util.function.ToDoubleFunction; import java.util.function.ToIntFunction; import java.util.function.ToLongFunction; import java.util.function.UnaryOperator; // @@@ Specification to-do list @@@ // - Describe the difference between sequential and parallel streams // - More general information about reduce, better definitions for associativity, more description of // how reduce employs parallelism, more examples // - Role of stream flags in various operations, specifically ordering // - Whether each op preserves encounter order // @@@ Specification to-do list @@@ /** * A sequence of elements supporting sequential and parallel bulk operations. * Streams support lazy intermediate operations (transforming a stream to * another stream) such as {@code filter} and {@code map}, and terminal * operations (consuming the contents of a stream to produce a result or * side-effect), such as {@code forEach}, {@code findFirst}, and {@code * iterator}. Once an operation has been performed on a stream, it * is considered consumed and no longer usable for other operations. * *

For sequential stream pipelines, all operations are performed in the * encounter order of the pipeline * source, if the pipeline source has a defined encounter order. * *

For parallel stream pipelines, unless otherwise specified, intermediate * stream operations preserve the * encounter order of their source, and terminal operations * respect the encounter order of their source, if the source * has an encounter order. Provided that and parameters to stream operations * satisfy the non-interference * requirements, and excepting differences arising from the absence of * a defined encounter order, the result of a stream pipeline should be the * stable across multiple executions of the same operations on the same source. * However, the timing and thread in which side-effects occur (for those * operations which are allowed to produce side-effects, such as * {@link #forEach(Consumer)}), are explicitly nondeterministic for parallel * execution of stream pipelines. * *

Unless otherwise noted, passing a {@code null} argument to any stream * method may result in a {@link NullPointerException}. * * @apiNote * Streams are not data structures; they do not manage the storage for their * elements, nor do they support access to individual elements. However, * you can use the {@link #iterator()} or {@link #spliterator()} operations to * perform a controlled traversal. * * @param type of elements * @since 1.8 * @see java.util.stream */ public interface Stream extends BaseStream> { /** * Returns a stream consisting of the elements of this stream that match * the given predicate. * *

This is an intermediate * operation. * * @param predicate a * non-interfering, stateless predicate to apply to * each element to determine if it should be included * @return the new stream */ Stream filter(Predicate predicate); /** * Returns a stream consisting of the results of applying the given * function to the elements of this stream. * *

This is an intermediate * operation. * * @param The element type of the new stream * @param mapper a * non-interfering, stateless function to apply to each * element * @return the new stream */ Stream map(Function mapper); /** * Returns an {@code IntStream} consisting of the results of applying the * given function to the elements of this stream. * *

This is an * intermediate operation. * * @param mapper a * non-interfering, stateless function to apply to each * element * @return the new stream */ IntStream mapToInt(ToIntFunction mapper); /** * Returns a {@code LongStream} consisting of the results of applying the * given function to the elements of this stream. * *

This is an intermediate * operation. * * @param mapper a * non-interfering, stateless function to apply to each * element * @return the new stream */ LongStream mapToLong(ToLongFunction mapper); /** * Returns a {@code DoubleStream} consisting of the results of applying the * given function to the elements of this stream. * *

This is an intermediate * operation. * * @param mapper a * non-interfering, stateless function to apply to each * element * @return the new stream */ DoubleStream mapToDouble(ToDoubleFunction mapper); /** * Returns a stream consisting of the results of replacing each element of * this stream with the contents of the stream produced by applying the * provided mapping function to each element. If the result of the mapping * function is {@code null}, this is treated as if the result is an empty * stream. * *

This is an intermediate * operation. * * @apiNote * The {@code flatMap()} operation has the effect of applying a one-to-many * tranformation to the elements of the stream, and then flattening the * resulting elements into a new stream. For example, if {@code orders} * is a stream of purchase orders, and each purchase order contains a * collection of line items, then the following produces a stream of line * items: *

{@code
     *     orderStream.flatMap(order -> order.getLineItems().stream())...
     * }
* * @param The element type of the new stream * @param mapper a * non-interfering, stateless function to apply to each * element which produces a stream of new values * @return the new stream */ Stream flatMap(Function> mapper); /** * Returns an {@code IntStream} consisting of the results of replacing each * element of this stream with the contents of the stream produced by * applying the provided mapping function to each element. If the result of * the mapping function is {@code null}, this is treated as if the result is * an empty stream. * *

This is an intermediate * operation. * * @param mapper a * non-interfering, stateless function to apply to each * element which produces a stream of new values * @return the new stream */ IntStream flatMapToInt(Function mapper); /** * Returns a {@code LongStream} consisting of the results of replacing each * element of this stream with the contents of the stream produced * by applying the provided mapping function to each element. If the result * of the mapping function is {@code null}, this is treated as if the * result is an empty stream. * *

This is an intermediate * operation. * * @param mapper a * non-interfering, stateless function to apply to * each element which produces a stream of new values * @return the new stream */ LongStream flatMapToLong(Function mapper); /** * Returns a {@code DoubleStream} consisting of the results of replacing each * element of this stream with the contents of the stream produced * by applying the provided mapping function to each element. If the result * of the mapping function is {@code null}, this is treated as if the result * is an empty stream. * *

This is an intermediate * operation. * * @param mapper a * non-interfering, stateless function to apply to each * element which produces a stream of new values * @return the new stream */ DoubleStream flatMapToDouble(Function mapper); /** * Returns a stream consisting of the distinct elements (according to * {@link Object#equals(Object)}) of this stream. * *

This is a stateful * intermediate operation. * * @return the new stream */ Stream distinct(); /** * Returns a stream consisting of the elements of this stream, sorted * according to natural order. If the elements of this stream are not * {@code Comparable}, a {@code java.lang.ClassCastException} may be thrown * when the stream pipeline is executed. * *

This is a stateful * intermediate operation. * * @return the new stream */ Stream sorted(); /** * Returns a stream consisting of the elements of this stream, sorted * according to the provided {@code Comparator}. * *

This is a stateful * intermediate operation. * * @param comparator a * non-interfering, stateless {@code Comparator} to * be used to compare stream elements * @return the new stream */ Stream sorted(Comparator comparator); /** * Returns a stream consisting of the elements of this stream, additionally * performing the provided action on each element as elements are consumed * from the resulting stream. * *

This is an intermediate * operation. * *

For parallel stream pipelines, the action may be called at * whatever time and in whatever thread the element is made available by the * upstream operation. If the action modifies shared state, * it is responsible for providing the required synchronization. * * @apiNote This method exists mainly to support debugging, where you want * to see the elements as they flow past a certain point in a pipeline: *

{@code
     *     list.stream()
     *         .filter(filteringFunction)
     *         .peek(e -> {System.out.println("Filtered value: " + e); });
     *         .map(mappingFunction)
     *         .peek(e -> {System.out.println("Mapped value: " + e); });
     *         .collect(Collectors.intoList());
     * }
* * @param consumer a * non-interfering action to perform on the elements as * they are consumed from the stream * @return the new stream */ Stream peek(Consumer consumer); /** * Returns a stream consisting of the elements of this stream, truncated * to be no longer than {@code maxSize} in length. * *

This is a short-circuiting * stateful intermediate operation. * * @param maxSize the number of elements the stream should be limited to * @return the new stream * @throws IllegalArgumentException if {@code maxSize} is negative */ Stream limit(long maxSize); /** * Returns a stream consisting of the remaining elements of this stream * after indexing {@code startInclusive} elements into the stream. If the * {@code startInclusive} index lies past the end of this stream then an * empty stream will be returned. * *

This is a stateful * intermediate operation. * * @param startInclusive the number of leading elements to skip * @return the new stream * @throws IllegalArgumentException if {@code startInclusive} is negative */ Stream substream(long startInclusive); /** * Returns a stream consisting of the remaining elements of this stream * after indexing {@code startInclusive} elements into the stream and * truncated to contain no more than {@code endExclusive - startInclusive} * elements. If the {@code startInclusive} index lies past the end * of this stream then an empty stream will be returned. * *

This is a short-circuiting * stateful intermediate operation. * * @param startInclusive the starting position of the substream, inclusive * @param endExclusive the ending position of the substream, exclusive * @return the new stream * @throws IllegalArgumentException if {@code startInclusive} or * {@code endExclusive} is negative or {@code startInclusive} is greater * than {@code endExclusive} */ Stream substream(long startInclusive, long endExclusive); /** * Performs an action for each element of this stream. * *

This is a terminal * operation. * *

For parallel stream pipelines, this operation does not * guarantee to respect the encounter order of the stream, as doing so * would sacrifice the benefit of parallelism. For any given element, the * action may be performed at whatever time and in whatever thread the * library chooses. If the action accesses shared state, it is * responsible for providing the required synchronization. * * @param action a * non-interfering action to perform on the elements */ void forEach(Consumer action); /** * Performs an action for each element of this stream, guaranteeing that * each element is processed in encounter order for streams that have a * defined encounter order. * *

This is a terminal * operation. * * @param action a * non-interfering action to perform on the elements * @see #forEach(Consumer) */ void forEachOrdered(Consumer action); /** * Returns an array containing the elements of this stream. * *

This is a terminal * operation. * * @return an array containing the elements of this stream */ Object[] toArray(); /** * Returns an array containing the elements of this stream, using the * provided {@code generator} function to allocate the returned array. * *

This is a terminal * operation. * * @param the element type of the resulting array * @param generator a function which produces a new array of the desired * type and the provided length * @return an array containing the elements in this stream * @throws ArrayStoreException if the runtime type of the array returned * from the array generator is not a supertype of the runtime type of every * element in this stream */ A[] toArray(IntFunction generator); /** * Performs a reduction on the * elements of this stream, using the provided identity value and an * associative * accumulation function, and returns the reduced value. This is equivalent * to: *

{@code
     *     T result = identity;
     *     for (T element : this stream)
     *         result = accumulator.apply(result, element)
     *     return result;
     * }
* * but is not constrained to execute sequentially. * *

The {@code identity} value must be an identity for the accumulator * function. This means that for all {@code t}, * {@code accumulator.apply(identity, t)} is equal to {@code t}. * The {@code accumulator} function must be an * associative function. * *

This is a terminal * operation. * * @apiNote Sum, min, max, average, and string concatenation are all special * cases of reduction. Summing a stream of numbers can be expressed as: * *

{@code
     *     Integer sum = integers.reduce(0, (a, b) -> a+b);
     * }
* * or more compactly: * *
{@code
     *     Integer sum = integers.reduce(0, Integer::sum);
     * }
* *

While this may seem a more roundabout way to perform an aggregation * compared to simply mutating a running total in a loop, reduction * operations parallelize more gracefully, without needing additional * synchronization and with greatly reduced risk of data races. * * @param identity the identity value for the accumulating function * @param accumulator an associative * non-interfering, * stateless function for combining two values * @return the result of the reduction */ T reduce(T identity, BinaryOperator accumulator); /** * Performs a reduction on the * elements of this stream, using an * associative accumulation * function, and returns an {@code Optional} describing the reduced value, * if any. This is equivalent to: *

{@code
     *     boolean foundAny = false;
     *     T result = null;
     *     for (T element : this stream) {
     *         if (!foundAny) {
     *             foundAny = true;
     *             result = element;
     *         }
     *         else
     *             result = accumulator.apply(result, element);
     *     }
     *     return foundAny ? Optional.of(result) : Optional.empty();
     * }
* * but is not constrained to execute sequentially. * *

The {@code accumulator} function must be an * associative function. * *

This is a terminal * operation. * * @param accumulator an associative * non-interfering, * stateless function for combining two values * @return the result of the reduction * @see #reduce(Object, BinaryOperator) * @see #min(java.util.Comparator) * @see #max(java.util.Comparator) */ Optional reduce(BinaryOperator accumulator); /** * Performs a reduction on the * elements of this stream, using the provided identity, accumulation * function, and a combining functions. This is equivalent to: *

{@code
     *     U result = identity;
     *     for (T element : this stream)
     *         result = accumulator.apply(result, element)
     *     return result;
     * }
* * but is not constrained to execute sequentially. * *

The {@code identity} value must be an identity for the combiner * function. This means that for all {@code u}, {@code combiner(identity, u)} * is equal to {@code u}. Additionally, the {@code combiner} function * must be compatible with the {@code accumulator} function; for all * {@code u} and {@code t}, the following must hold: *

{@code
     *     combiner.apply(u, accumulator.apply(identity, t)) == accumulator.apply(u, t)
     * }
* *

This is a terminal * operation. * * @apiNote Many reductions using this form can be represented more simply * by an explicit combination of {@code map} and {@code reduce} operations. * The {@code accumulator} function acts as a fused mapper and accumulator, * which can sometimes be more efficient than separate mapping and reduction, * such as in the case where knowing the previously reduced value allows you * to avoid some computation. * * @param The type of the result * @param identity the identity value for the combiner function * @param accumulator an associative * non-interfering, * stateless function for incorporating an additional * element into a result * @param combiner an associative * non-interfering, * stateless function for combining two values, which * must be compatible with the accumulator function * @return the result of the reduction * @see #reduce(BinaryOperator) * @see #reduce(Object, BinaryOperator) */ U reduce(U identity, BiFunction accumulator, BinaryOperator combiner); /** * Performs a mutable * reduction operation on the elements of this stream. A mutable * reduction is one in which the reduced value is a mutable value holder, * such as an {@code ArrayList}, and elements are incorporated by updating * the state of the result, rather than by replacing the result. This * produces a result equivalent to: *

{@code
     *     R result = resultFactory.get();
     *     for (T element : this stream)
     *         accumulator.accept(result, element);
     *     return result;
     * }
* *

Like {@link #reduce(Object, BinaryOperator)}, {@code collect} operations * can be parallelized without requiring additional synchronization. * *

This is a terminal * operation. * * @apiNote There are many existing classes in the JDK whose signatures are * a good match for use as arguments to {@code collect()}. For example, * the following will accumulate strings into an ArrayList: *

{@code
     *     List asList = stringStream.collect(ArrayList::new, ArrayList::add, ArrayList::addAll);
     * }
* *

The following will take a stream of strings and concatenates them into a * single string: *

{@code
     *     String concat = stringStream.collect(StringBuilder::new, StringBuilder::append,
     *                                          StringBuilder::append)
     *                                 .toString();
     * }
* * @param type of the result * @param resultFactory a function that creates a new result container. * For a parallel execution, this function may be * called multiple times and must return a fresh value * each time. * @param accumulator an associative * non-interfering, * stateless function for incorporating an additional * element into a result * @param combiner an associative * non-interfering, * stateless function for combining two values, which * must be compatible with the accumulator function * @return the result of the reduction */ R collect(Supplier resultFactory, BiConsumer accumulator, BiConsumer combiner); /** * Performs a mutable * reduction operation on the elements of this stream using a * {@code Collector} object to describe the reduction. A {@code Collector} * encapsulates the functions used as arguments to * {@link #collect(Supplier, BiConsumer, BiConsumer)}, allowing for reuse of * collection strategies, and composition of collect operations such as * multiple-level grouping or partitioning. * *

This is a terminal * operation. * *

When executed in parallel, multiple intermediate results may be * instantiated, populated, and merged, so as to maintain isolation of * mutable data structures. Therefore, even when executed in parallel * with non-thread-safe data structures (such as {@code ArrayList}), no * additional synchronization is needed for a parallel reduction. * * @apiNote * The following will accumulate strings into an ArrayList: *

{@code
     *     List asList = stringStream.collect(Collectors.toList());
     * }
* *

The following will classify {@code Person} objects by city: *

{@code
     *     Map> peopleByCity
     *         = personStream.collect(Collectors.groupBy(Person::getCity));
     * }
* *

The following will classify {@code Person} objects by state and city, * cascading two {@code Collector}s together: *

{@code
     *     Map>> peopleByStateAndCity
     *         = personStream.collect(Collectors.groupBy(Person::getState,
     *                                                   Collectors.groupBy(Person::getCity)));
     * }
* * @param the type of the result * @param 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 collect(Collector collector); /** * Returns the minimum element of this stream according to the provided * {@code Comparator}. This is a special case of a * reduction. * *

This is a terminal operation. * * @param comparator a non-interfering, * stateless {@code Comparator} to use to compare * elements of this stream * @return an {@code Optional} describing the minimum element of this stream, * or an empty {@code Optional} if the stream is empty */ Optional min(Comparator comparator); /** * Returns the maximum element of this stream according to the provided * {@code Comparator}. This is a special case of a * reduction. * *

This is a terminal * operation. * * @param comparator a non-interfering, * stateless {@code Comparator} to use to compare * elements of this stream * @return an {@code Optional} describing the maximum element of this stream, * or an empty {@code Optional} if the stream is empty */ Optional max(Comparator comparator); /** * Returns the count of elements in this stream. This is a special case of * a reduction and is * equivalent to: *

{@code
     *     return mapToLong(e -> 1L).sum();
     * }
* *

This is a terminal operation. * * @return the count of elements in this stream */ long count(); /** * Returns whether any elements of this stream match the provided * predicate. May not evaluate the predicate on all elements if not * necessary for determining the result. * *

This is a short-circuiting * terminal operation. * * @param predicate a non-interfering, * stateless predicate to apply to elements of this * stream * @return {@code true} if any elements of the stream match the provided * predicate otherwise {@code false} */ boolean anyMatch(Predicate predicate); /** * Returns whether all elements of this stream match the provided predicate. * May not evaluate the predicate on all elements if not necessary for * determining the result. * *

This is a short-circuiting * terminal operation. * * @param predicate a non-interfering, * stateless predicate to apply to elements of this * stream * @return {@code true} if all elements of the stream match the provided * predicate otherwise {@code false} */ boolean allMatch(Predicate predicate); /** * Returns whether no elements of this stream match the provided predicate. * May not evaluate the predicate on all elements if not necessary for * determining the result. * *

This is a short-circuiting * terminal operation. * * @param predicate a non-interfering, * stateless predicate to apply to elements of this * stream * @return {@code true} if no elements of the stream match the provided * predicate otherwise {@code false} */ boolean noneMatch(Predicate predicate); /** * Returns an {@link Optional} describing the first element of this stream * (in the encounter order), or an empty {@code Optional} if the stream is * empty. If the stream has no encounter order, then any element may be * returned. * *

This is a short-circuiting * terminal operation. * * @return an {@code Optional} describing the first element of this stream, * or an empty {@code Optional} if the stream is empty * @throws NullPointerException if the element selected is null */ Optional findFirst(); /** * Returns an {@link Optional} describing some element of the stream, or an * empty {@code Optional} if the stream is empty. * *

This is a short-circuiting * terminal operation. * *

The behavior of this operation is explicitly nondeterministic; it is * free to select any element in the stream. This is to allow for maximal * performance in parallel operations; the cost is that multiple invocations * on the same source may not return the same result. (If the first element * in the encounter order is desired, use {@link #findFirst()} instead.) * * @return an {@code Optional} describing some element of this stream, or an * empty {@code Optional} if the stream is empty * @throws NullPointerException if the element selected is null * @see #findFirst() */ Optional findAny(); // Static factories /** * Returns a builder for a {@code Stream}. * * @param type of elements * @return a stream builder */ public static StreamBuilder builder() { return new Streams.StreamBuilderImpl<>(); } /** * Returns an empty sequential {@code Stream}. * * @param the type of stream elements * @return an empty sequential stream */ public static Stream empty() { return StreamSupport.stream(Spliterators.emptySpliterator()); } /** * Returns a sequential {@code Stream} containing a single element. * * @param t the single element * @param the type of stream elements * @return a singleton sequential stream */ public static Stream of(T t) { return StreamSupport.stream(new Streams.StreamBuilderImpl<>(t)); } /** * Returns a sequential stream whose elements are the specified values. * * @param the type of stream elements * @param values the elements of the new stream * @return the new stream */ @SafeVarargs public static Stream of(T... values) { return Arrays.stream(values); } /** * Returns an infinite sequential {@code Stream} produced by iterative * application of a function {@code f} to an initial element {@code seed}, * producing a {@code Stream} consisting of {@code seed}, {@code f(seed)}, * {@code f(f(seed))}, etc. * *

The first element (position {@code 0}) in the {@code Stream} will be * the provided {@code seed}. For {@code n > 0}, the element at position * {@code n}, will be the result of applying the function {@code f} to the * element at position {@code n - 1}. * * @param the type of stream elements * @param seed the initial element * @param f a function to be applied to to the previous element to produce * a new element * @return a new sequential {@code Stream} */ public static Stream iterate(final T seed, final UnaryOperator f) { Objects.requireNonNull(f); final Iterator iterator = new Iterator() { @SuppressWarnings("unchecked") T t = (T) Streams.NONE; @Override public boolean hasNext() { return true; } @Override public T next() { return t = (t == Streams.NONE) ? seed : f.apply(t); } }; return StreamSupport.stream(Spliterators.spliteratorUnknownSize( iterator, Spliterator.ORDERED | Spliterator.IMMUTABLE)); } /** * Returns a sequential {@code Stream} where each element is * generated by a {@code Supplier}. This is suitable for generating * constant streams, streams of random elements, etc. * * @param the type of stream elements * @param s the {@code Supplier} of generated elements * @return a new sequential {@code Stream} */ public static Stream generate(Supplier s) { Objects.requireNonNull(s); return StreamSupport.stream( new StreamSpliterators.InfiniteSupplyingSpliterator.OfRef<>(Long.MAX_VALUE, s)); } /** * Creates a lazy concatenated {@code Stream} whose elements are all the * elements of a first {@code Stream} succeeded by all the elements of the * second {@code Stream}. The resulting stream is ordered if both * of the input streams are ordered, and parallel if either of the input * streams is parallel. * * @param The type of stream elements * @param a the first stream * @param b the second stream to concatenate on to end of the first * stream * @return the concatenation of the two input streams */ public static Stream concat(Stream a, Stream b) { Objects.requireNonNull(a); Objects.requireNonNull(b); @SuppressWarnings("unchecked") Spliterator split = new Streams.ConcatSpliterator.OfRef<>( (Spliterator) a.spliterator(), (Spliterator) b.spliterator()); return (a.isParallel() || b.isParallel()) ? StreamSupport.parallelStream(split) : StreamSupport.stream(split); } }