RX - Notifications at a specified rate

Question

I'm a newbie with RX and I'm facing a problem with "shaping the notifications traffic".

I wonder how can I notify observers with a given throughput; that is, I would like that "OnNext" method is called not before a given amount of time is elapsed since the last "OnNext" invocation.

For the sake of completeness: I want that every element in the sequence will be notified.

For example, with a 0.2 symbols/tick:

Tick:        0         10        20        30
             |---------|---------|---------|
Producer:    A---B------C--D-----E-------F
Result:      A    B     C    D    E      F
             0    5     11   16   21     28

Is there a way to compose the observable or I have to implement my own Subject?

Thanks a lot

Answer 1

yeah just turn each value into an async process that does not complete until delay has elapsed and then concatenate them.

var delay = Observable.Empty<T>().Delay(TimeSpan.FromSeconds(2));
var rateLimited = source
    .Select(item => Observable.Return(item).Concat(delay))
    .Concat();