given an infinite sequence break it into intervals, and return a new infinite sequence with the average of each interval

Question

i have to calculate the average of a Infinite Sequence using Stream API

Input:

Stream<Double> s = a,b,c,d ...
int interval = 3

Expected Result:

Stream<Double> result = avg(a,b,c), avg(d,e,f), ....

the result can be also an Iterator, or any other type as long as it mantains the structure of an infinite list

of course what i written is pseudo code and doesnt run

Answer 1

There is a @Beta API termed mapWithIndex within Guava that could help here with certain assumption:

static Stream<Double> stepAverage(Stream<Double> stream, int step) {
    return Streams.mapWithIndex(stream, (from, index) -> Map.entry(index, from))
            .collect(Collectors.groupingBy(e -> (e.getKey() / step), TreeMap::new,
                    Collectors.averagingDouble(Map.Entry::getValue)))
            .values().stream();
}

The assumption that it brings in is detailed in the documentation clearly(emphasized by me):

The resulting stream is efficiently splittable if and only if stream was efficiently splittable and its underlying spliterator reported Spliterator.SUBSIZED. This is generally the case if the underlying stream comes from a data structure supporting efficient indexed random access, typically an array or list.

Answer 2

This should work fine using vanilla Java

I'm using Stream#mapMulti and a Set external to the Stream to aggregate the doubles

As you see, I also used DoubleSummaryStatistics to count the average. I could have use the traditional looping and summing then dividing but I found this way more explicit

Update: I changed the Collection used from Set to List as a Set could cause unexpected behaviour

int step = 3;
List<Double> list = new ArrayList<>();
Stream<Double> averagesStream =
        infiniteStream.mapMulti(((Double aDouble, Consumer<Double> doubleConsumer) -> {
            list.add(aDouble);
            if (list.size() == step) {
                DoubleSummaryStatistics doubleSummaryStatistics = new DoubleSummaryStatistics();
                list.forEach(doubleSummaryStatistics::accept);
                list.clear();
                doubleConsumer.accept(doubleSummaryStatistics.getAverage());
            }
        }));