Package com.google.mu.util.stream

Class MoreStreams

java.lang.Object
com.google.mu.util.stream.MoreStreams
public final class MoreStreams extends Object
Static utilities pertaining to Stream in addition to relevant utilities in JDK and Guava.
Since:
1.1

  • Method Details

    • generate

      public static <T> Stream<T> generate(T seed, Function<? super T,? extends Stream<? extends T>> step)
      Returns a Stream produced by iterative application of step to the initial seed, producing a Stream consisting of seed, elements of step(seed), elements of step(x) for each x in step(seed), etc. (If the result stream returned by the step function is null an empty stream is used, instead.)

      While Stream.generate(supplier) can be used to generate infinite streams, it's not as easy to generate a finite stream unless the size can be pre-determined. This method can be used to generate finite streams: just return an empty stream when the step determines that there's no more elements to be generated.

      A typical group of use cases are BFS traversal algorithms. For example, to stream the tree nodes in BFS order: 

      
   Stream<Node> bfs(Node root) {
     return generate(root, node -> node.children().stream());
   }
      It's functionally equivalent to the following common imperative code: 
      
   List<Node> bfs(Node root) {
     List<Node> result = new ArrayList<>();
     Queue<Node> queue = new ArrayDeque<>();
     queue.add(root);
     while (!queue.isEmpty()) {
       Node node = queue.remove();
       result.add(node);
       queue.addAll(node.children());
     }
     return result;
   }
      A BFS 2-D grid traversal algorithm: 
      
   Stream<Cell> bfs(Cell startingCell) {
     Set<Cell> visited = new HashSet<>();
     visited.add(startingCell);
     return generate(startingCell, c -> c.neighbors().filter(visited::add));
   }

      At every step, 0, 1 or more elements can be generated into the resulting stream. As discussed above, returning an empty stream leads to eventual termination of the stream; returning 1-element stream is equivalent to Stream.generate(supplier); while returning more than one elements allows a single element to fan out to multiple elements.

      Since:
      1.9

    • flatten

      @Deprecated public static <T> Stream<T> flatten(Stream<? extends Stream<? extends T>> streamOfStream)
      Deprecated.
      Use flatMap() in Java 10+
      Flattens streamOfStream and returns an unordered sequential stream of the nested elements.

      Logically, stream.flatMap(fanOut) is equivalent to MoreStreams.flatten(stream.map(fanOut)). Due to this JDK bug, flatMap() uses forEach() internally and doesn't support short-circuiting for the passed-in stream. flatten() supports short-circuiting and can be used to flatten infinite streams.

      Since:
      1.9

    • groupConsecutive

      public static <T, R> Stream<R> groupConsecutive(Stream<T> stream, BiPredicate<? super T,? super T> sameGroup, Collector<? super T,?,R> groupCollector)
      Groups consecutive elements from stream lazily. Two consecutive elements belong to the same group if sameGroup evaluates to true. Consecutive elements belonging to the same group will be collected together using groupCollector.

      For example, you can find every list of increasing stock prices, given daily stock prices: 

      
 ImmutableList<ImmutableList<Double>> increasingStockPriceSeries =
     groupConsecutive(stockPrices, (p1, p2) -> p1 <= p2, toImmutableList())
         .collect(toImmutableList());
      Since:
      5.7

    • groupConsecutive

      public static <T> Stream<T> groupConsecutive(Stream<T> stream, BiPredicate<? super T,? super T> sameGroup, BinaryOperator<T> groupReducer)
      Groups consecutive elements from stream lazily. Two consecutive elements belong to the same group if sameGroup evaluates to true. Consecutive elements belonging to the same group will be reduced using groupReducer.

      For example, you can find the total number of trades for the stock during each period when there was no large trade anomaly (difference): 

      
 ImmutableList<Long> stockTradesPerPeriod =
     groupConsecutive(stockTrades, (t1, t2) -> Math.abs(t1 - t2) < threshold, Long::sum)
         .collect(toImmutableList());
      Since:
      5.7

    • groupConsecutive

      public static <T, R> Stream<R> groupConsecutive(Stream<T> stream, Function<? super T,?> groupKeyFunction, Collector<? super T,?,R> groupCollector)
      Groups consecutive elements from stream lazily. Two consecutive elements belong to the same group if groupKeyFunction evaluates to equal keys. Consecutive elements belonging to the same group will be collected together using groupCollector.

      For example, you can group consecutive events by their severity: 

      
 ImmutableList<ImmutableList<Event>> sameSeverityEventGroups =
     groupConsecutive(events, Event::severity, toImmutableList())
         .collect(toImmutableList());
      Since:
      5.7

    • groupConsecutive

      public static <T> Stream<T> groupConsecutive(Stream<T> stream, Function<? super T,?> groupKeyFunction, BinaryOperator<T> groupReducer)
      Groups consecutive elements from stream lazily. Two consecutive elements belong to the same group if groupKeyFunction evaluates to equal keys. Consecutive elements belonging to the same group will be reduced using groupReducer.

      For example, you can find the first event of each severity in a consecutive series of events: 

      
 ImmutableList<Event> firstEventsWithAlternatingSeverity =
     groupConsecutive(events, Event::severity, (e1, e2) -> e1)
         .collect(toImmutableList());
      Since:
      5.7

    • runLengthEncode

      public static <T> BiStream<T,Long> runLengthEncode(Stream<T> stream, BiPredicate<? super T,? super T> sameGroup)
      Groups consecutive items in stream using the sameGroup predicate, along with the group's run length (number of items).

      The following example encodes a stream of payloads with run length: 

      
 ImmutableList<RunLengthEncodedPayload> encodedPayloads =
     runLengthEncode(payloads.stream(), payloadDiffer::isEquivalent)
         .mapToObj(
             (payload, count) ->
                 RunLengthEncodedPayload.newBuilder()
                     .setPayload(payload)
                     .setCount(count)
                     .build())
         .collect(toImmutableList());
      Returns:
      a BiStream with the first item of each group and the run length of that group.
      Since:
      7.0

    • iterateOnce

      public static <T> Iterable<T> iterateOnce(Stream<T> stream)
      Iterates through stream only once. It's strongly recommended to avoid assigning the return value to a variable or passing it to any other method because the returned Iterable's iterator() method can only be called once. Instead, always use it together with a for-each loop, as in: 
      
   for (Foo foo : iterateOnce(stream)) {
     ...
     if (...) continue;
     if (...) break;
     ...
   }
      The above is equivalent to manually doing: 
      
   Iterable<Foo> foos = stream::iterator;
   for (Foo foo : foos) {
     ...
   }
      except using this API eliminates the need for a named variable that escapes the scope of the for-each loop. And code is more readable too.

      Note that iterateThrough() should be preferred whenever possible due to the caveats mentioned above. This method is still useful when the loop body needs to use control flows such as break or return.

    • iterateThrough

      public static <T, E extends Throwable> void iterateThrough(Stream<? extends T> stream, CheckedConsumer<? super T,E> consumer) throws E
      Iterates through stream sequentially and passes each element to consumer with exceptions propagated. For example: 
      
   void writeAll(Stream<?> stream, ObjectOutput out) throws IOException {
     iterateThrough(stream, out::writeObject);
   }
      Throws:
      E

    • dice

      public static <T> Stream<List<T>> dice(Stream<? extends T> stream, int maxSize)
      Dices stream into smaller chunks each with up to maxSize elements.

      For a sequential stream, the first N-1 chunk's will contain exactly maxSize elements and the last chunk may contain less (but never 0). However for parallel streams, it's possible that the stream is split in roughly equal-sized sub streams before being diced into smaller chunks, which then will result in more than one chunks with less than maxSize elements.

      This is an intermediary operation.

      Parameters:
      stream - the source stream to be diced
      maxSize - the maximum size for each chunk
      Returns:
      Stream of diced chunks each being a list of size up to maxSize
      Throws:
      IllegalStateException - if maxSize <= 0

    • dice

      public static <T> Spliterator<List<T>> dice(Spliterator<? extends T> spliterator, int maxSize)
      Dices spliterator into smaller chunks each with up to maxSize elements.
      Parameters:
      spliterator - the source spliterator to be diced
      maxSize - the maximum size for each chunk
      Returns:
      Spliterator of diced chunks each being a list of size up to maxSize
      Throws:
      IllegalStateException - if maxSize <= 0

    • indexesFrom

      public static Stream<Integer> indexesFrom(int firstIndex)
      Returns an infinite Stream starting from firstIndex. Can be used together with BiStream.zip(java.util.Collection<L>, java.util.Collection<R>) to iterate over a stream with index. For example: zip(indexesFrom(0), values).

      To get a finite stream, use indexesFrom(...).limit(size).

      Note that while indexesFrom(0) will eventually incur boxing cost for every integer, the JVM typically pre-caches small Integer instances (by default up to 127).

      Since:
      3.7

    • indexesFrom

      public static Stream<Long> indexesFrom(long firstIndex)
      Returns an infinite index stream starting from firstIndex. This can then be used to zip with another stream to provide indexing, such as: 
         BiStream.zip(indexesFrom(0), values).toMap();

      To get a finite stream, use indexesFrom(0).limit(size).

      For small indexes (up to 127), Long instances are pre-cached by JVM so no boxing happens; for larger indexes, every index incurs some boxing cost. If the cost is of concern, prefer to use LongStream.iterate(long, java.util.function.LongUnaryOperator) directly.

      Since:
      5.7

    • whileNotNull

      public static <T> Stream<T> whileNotNull(Supplier<? extends T> supplier)
      Similar to Stream.generate(java.util.function.Supplier<? extends T>), returns an infinite, sequential, ordered, and non-null stream where each element is generated by the provided Supplier. The stream however will terminate as soon as the Supplier returns null, in which case the null is treated as the terminal condition and doesn't constitute a stream element.

      For sequential iterations, whileNotNll() is usually more concise than implementing Spliterators.AbstractSpliterator directly. The latter requires boilerplate that looks like this: 

      
 return StreamSupport.stream(
     new AbstractSpliterator<T>(MAX_VALUE, NONNULL | ORDERED) {
       public boolean tryAdvance(Consumer<? super T> action) {
         if (hasData) {
           action.accept(data);
           return true;
         }
         return false;
       }
     }, false);
      Which is equivalent to the following one-liner using whileNotNull(): 
      
 return whileNotNull(() -> hasData ? data : null);

      Why null? Why not Optional? Wrapping every generated element of a stream in an Optional carries considerable allocation cost. Also, while nulls are in general discouraged, they are mainly a problem for users who have to remember to deal with them. The stream returned by whileNotNull() on the other hand is guaranteed to never include nulls that users have to worry about.

      If you already have an Optional from a method return value, you can use whileNotNull(() -> optionalReturningMethod().orElse(null)).

      One may still need to implement AbstractSpliterator or Iterator directly if null is a valid element (usually discouraged though).

      If you have an imperative loop over a mutable queue or stack: 

      
 while (!queue.isEmpty()) {
   int num = queue.poll();
   if (someCondition) {
     ...
   }
 }
      it can be turned into a stream using whileNotNull(): 
      
 whileNotNull(queue::poll).filter(someCondition)...
      Since:
      4.1

    • withSideEffect

      public static <T> Stream<T> withSideEffect(Stream<T> stream, Consumer<? super T> sideEffect)
      Returns a sequential stream with sideEfect attached on every element.

      Unlike Stream.peek(java.util.function.Consumer<? super T>), which should only be used for debugging purpose, the side effect is allowed to interfere with the source of the stream, and is guaranteed to be applied in encounter order.

      If you have to resort to side effects, use this dedicated method instead of peek() or any other stream method. From the API specification, all methods defined by Stream are expected to be stateless, and should not cause or depend on side effects, because even for ordered, sequential streams, only the order of output is defined, not the order of evaluation.

      Since:
      4.9