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, Apr 25 '15 at 00:15
There are many more Rx frameworks now for a myriad of languages. I don't know how similar they all are. — Enigmativity, Apr 25 '15 at 00:35

score 8 · Accepted Answer · answered Apr 25 '15 at 00:37

8

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.

answered Apr 25 '15 at 00:37

Enigmativity

113,464
11
89
172

I tried with multiple values to validate and it works exactly as expected! Thanks!! – Absolom Apr 25 '15 at 01:04
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 Apr 25 '15 at 01:42
1

@Absolom - I don't think it would lose any values as each operator is working from a single sequence. – Enigmativity Apr 25 '15 at 04:40

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

1 Answers1

Linked