Take every Nth element from an Observable

user_401828 0
Viewed 1989

I need to observe a stream that produces values at a regular interval, but there are too many values so I want to take, say one value for every ten values produced.

The equivalent, non-RX way to do that, would be:

int step = 10;  // take every 10th value
var numbers = Enumerable.Range(0, 100).Where((e, i) => i % step == 0);

What would be the idiomatic way to do that in Rx?

system.reactive c#
2 Answers

I would avoid using the Where operator taking the Func<TSource, int, bool> predicate parameter, out of fear of a possible OverflowException. This could happen in case the source sequence produces more than 2,147,483,647 elements. Both Linq and Rx operators allow arithmetic overflows to occur, and I am not aware of any workaround that prevents this from happening.

Here is a custom TakeEvery operator that is safe in that regard:

/// <summary>Samples the observable sequence by returning one every N elements,
/// ignoring the other elements.</summary>
public static IObservable<T> TakeEvery<T>(this IObservable<T> source, int step)
{
    if (step < 1) throw new ArgumentOutOfRangeException(nameof(step));
    return Observable.Defer(() =>
    {
        int index = -1;
        return source.Where(value => (index = ++index % step) == 0);
    });
}
user_111785490
Related