RX - Group/Batch bursts of elements in an observable sequence

Question

I have an observable sequence. When the first element is inserted, I would like to start a timer and batch subsequent inserted elements during the timespan of the timer. Then, the timer wouldn't start again until another element is inserted in the sequence.

So something like this:

--------|=====timespan====|---------------|=====timespan====|--------------> 1 2 3 4 5 6 7 8

would produce:

[1,2,3,4,5], [6,7,8]

I tried with Observable.Buffer() and a timespan but from my experimentation, I can see that the timer is started as soon as we subscribe to the observable sequence and is restarted as soon as the previous timer is completed.

So having the same sequence as the previous example and using the Buffer() with a timespan, I would have something like this:

|=====timespan====|=====timespan====|=====timespan====|=====timespan====|--> 1 2 3 4 5 6 7 8

which would produce this:

[1,2,3,4], [5], [6,7], [8]

Here is how I tested this behavior with the Buffer:

var source = Observable.Concat(Observable.Timer(TimeSpan.FromSeconds(6)).Select(o => 1), Observable.Timer(TimeSpan.FromSeconds(1)).Select(o => 2), Observable.Timer(TimeSpan.FromSeconds(3)).Select(o => 3), Observable.Never<int>()); Console.WriteLine("{0} => Started", DateTime.Now); source.Buffer(TimeSpan.FromSeconds(4)) .Subscribe(i => Console.WriteLine("{0} => [{1}]", DateTime.Now, string.Join(",", i)));

With the output:

4/24/2015 7:01:09 PM => Started 4/24/2015 7:01:13 PM => [] 4/24/2015 7:01:17 PM => [1,2] 4/24/2015 7:01:21 PM => [3] 4/24/2015 7:01:25 PM => [] 4/24/2015 7:01:29 PM => [] 4/24/2015 7:01:33 PM => []

Anyone has an idea on how to do this? Thanks in advance!

I am guessing that the logic would be the same in Java or in C#. If not, please let me know and I will remove the tag. Thanks! — Absolom
– Absolom, Commented Apr 25, 2015 at 0:15
There are many more Rx frameworks now for a myriad of languages. I don't know how similar they all are. — Enigmativity
– Enigmativity, Commented Apr 25, 2015 at 0:35

Enigmativity · Accepted Answer · 2015-04-25 00:37:43Z

Give this a go:

var source = Observable.Concat(Observable.Timer(TimeSpan.FromSeconds(6)).Select(o => 1), Observable.Timer(TimeSpan.FromSeconds(1)).Select(o => 2), Observable.Timer(TimeSpan.FromSeconds(4)).Select(o => 3), Observable.Never<int>()); Console.WriteLine("{0} => Started", DateTime.Now); source .GroupByUntil(x => 1, g => Observable.Timer(TimeSpan.FromSeconds(4))) .Select(x => x.ToArray()) .Switch() .Subscribe(i => Console.WriteLine("{0} => [{1}]", DateTime.Now, string.Join(",", i)));

I had to change your test code duration for the third timer to make sure the value was outside of the grouped timer.

I tried with multiple values to validate and it works exactly as expected! Thanks!!
Is there any possibility that data can be lost with this? The source sequence being generated by different threads (Observable.Timer), could there be any racing conditions?
@Absolom - I don't think it would lose any values as each operator is working from a single sequence.

Collectives™ on Stack Overflow

RX - Group/Batch bursts of elements in an observable sequence

1 Answer 1

3 Comments

Linked

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

3 Comments

Linked

Related