java - Create stream of streams from one long stream

Question

I want to split a single Stream into a Stream of Streams based on the contents of the Streams. The resulting the Stream should contain part of the original streams' data.

My real application is more complex (it is grouping log lines that are within a list of time intervals), but my problem is how to handle the streams, so here I ask about a simplified example.

Example problem

I want to be able split a Stream<Integer> into a Stream<Stream<Integer>> based on the same number being repeated, only leaving the streams with odd numbers.

For example the following stream containing:

{1,1,1,2,2,2,3,6,7,7,1,1}

Would need to result in a stream of streams containing:

{{1,1,1},{3},{7,7},{1,1}}

Leaving out the even numbers I can do by starting (or ending) with a filter:

Stream<Integer> input = ...;
Straem<Stream<Integer>> output = input.filter(this::isOdd).someOtherOperation();

This is undesired as it would mean evaluating each input value twice, this is acceptable but I would prefer avoiding this.

Ideas for solutions

My current solution does this iterating over the contents of the stream and creating a List<List<Integer>> and converting that to a Stream<Stream<Integer>>. However this means the full result is kept in memory (which is undesired for my application).

I also think I would be able to pull this of by writing my own Iterator that reads from the stream, but I am not sure how this would work.

Question

How can I convert a Stream into a Stream of Streams based on the contents of the original Stream, without storing the full result in as a List of Lists first.

Answer 1

You may want to implement your own aggregating spliterator to do this. There's already something similar in the proton-pack library (the first link redirects to the one implemented in proton-pack).

Note that you get a Stream<List<Integer>> (you may try to modify the implementation to have a Stream<Stream<Integer>> directly, but you always need to buffer a small amount elements; depending on the window's size; to test whether you should create a new window or not). So for example:

StreamUtils.aggregate(Stream.of(1, 1, 1, 2, 2, 2, 3, 6, 7, 7, 1, 1), 
                      Objects::equals)
           .forEach(System.out::println);

outputs:

[1, 1, 1]
[2, 2, 2]
[3]
[6]
[7, 7]
[1, 1]