Observable throttling

Question

I am trying to implement an event throttling with reactive extensions. I have a system where events might be raised with high frequency for a specific user or other entity type. What i need is to delay the event for specific amount of time and once the timeout expires raise the event with the last value.

What i have done is this

 private Subject<int> userBalanceObservable = new Subject<int>();
 userBalanceObservable.Sample(TimeSpan.FromSeconds(sampleSeconds))
            .Subscribe(sample => OnRaiseBalanceEvent(sample));

when event occurs

userBalanceObservable.OnNext(userId);

Edit

The problem with this approach is that the event is raised for the last value passed to OnNext, what i would actually need is to have a dellay for each value passed to OnNext.

E.g OnNext(1),OnNext(2),OnNext(3) i would need to have a delayed call for 1,2,3 instead i am getting only the last value which is 3.

Answer 1

Sample publishes the last value every time the sample interval is hit. If that is the behaviour you require then it is fine. Have a look at http://www.introtorx.com/content/v1.0.10621.0/13_TimeShiftedSequences.html#Sample For more information + other ways to slow down the emission rate.

In regards to the new information about your question:

If you want to emit all the values after a certain timeout has been hit you can group these until your timeout has hit (NOTE: You can run out of memory if you keep adding events without your timeout ever hitting due to the frequencey of event emissions)

You can create a Buffer which fills up until a Debounce timeout hits, see this answer on SO for pointers: How to implement buffering with timeout in RX