CODEWITHSUNDEEP
system.reactive
Soe Moe
Soe Moe

Reputation: 3438

data-based buffering in Rx

Let me explain what I want to achieve first.

Lets say I have the following data incoming form the event stream

var data = new string[] { 
                "hello", 
                "Using", 
                "ok:michael", 
                "ok", 
                "begin:events", 
                "1:232", 
                "2:343", 
                "end:events", 
                "error:dfljsdf",
                "fdl", 
                "error:fjkdjslf",
                "ok"  
            };

When I subscribe the data source, I would like to get the following result

"ok:michael"
"ok"
"begin:events 1:232 2:343 end:events"
"error:dfljsdf"
"error:fjkdjslf"
"ok"

Basically, I want to get whichever data that start with ok or error and the data between begin and end.

I have tried this so far..

var data = new string[] { 
                "hello", 
                "Using", 
                "ok:michael", 
                "ok", 
                "begin:events", 
                "1:232", 
                "2:343", 
                "end:events", 
                "error:dfljsdf",
                "fdl", 
                "error:fjkdjslf",
                "ok"  
            };



            var dataStream = Observable.Generate(
                                data.GetEnumerator(), 
                                e => e.MoveNext(), 
                                e => e, 
                                e => e.Current.ToString(), 
                                e => TimeSpan.FromSeconds(0.1));         

            var onelineStream = from d in dataStream
                                where d.StartsWith("ok") || d.StartsWith("error")
                                select d;

            // ???
            // may be need to buffer? I want to get data like "begin:events 1:232 2:343 end:events"
            // but it is not working...
            var multiLineStream = from list in dataStream.Buffer<string, string, string>(
                                bufferOpenings: dataStream.Where(d => d.StartsWith("begin")),
                                bufferClosingSelector: b => dataStream.Where(d => d.StartsWith("end")))
                              select String.Join(" ", list);

            // merge two stream????
            // but I have no clue how to merge these twos :(

            mergeStream .Subscribe(d =>
            {
                Console.WriteLine(d);
                Console.WriteLine();
            });

Since I'm very new to Reactive programming, I can't make myself to think in reactive way. :(

Thanks in advance.

Upvotes: 1

Views: 389

Answers (1)

Enigmativity
Enigmativity

Reputation: 117029

You were so, so very close to the right answer!

Essentially you had the onelineStream & multiLineStream queries just about right.

Merging them together is very easy. Just do this:

onelineStream.Merge(multiLineStream)

However, where your queries fell short was in the Observable.Generate that you used to introduce the delay between values. This creates a observable that, if you have multiple subscribers, kind of "fans out" the values.

Given your data and your definition for dataStream look how this code behaves:

dataStream.Select(x => "!" + x).Subscribe(Console.WriteLine);
dataStream.Select(x => "@" + x).Subscribe(Console.WriteLine);

You get these values:

!hello
@Using
!ok:michael
@ok
@1:232
!begin:events
@2:343
!end:events
!fdl
@error:dfljsdf
!error:fjkdjslf
@ok

Notice that some got handled by one subscription and the others got handled by the other. This means that even though your onelineStream & multiLineStream queries were just about right they would only see some of the data each and thus not behave as you expect.

You can also get race conditions that can skip and duplicate values. So it's best to avoid this kind of observable.

A better approach to introduce a delay between values is to do this:

var dataStream = data.ToObservable().Do(_ => Thread.Sleep(100));

Now this creates a "cold" observable, meaning that every new subscriber will get a fresh subscription of the observable so starting from the first value.

Your multiLineStream query will not work correctly on a cold observable.

To make the data stream a "hot" observable (which shares values amongst the subscribers) we use the Publish operator.

So, multiLineStream now looks like this:

var multiLineStream =
    dataStream.Publish(ds =>
        from list in ds.Buffer(
            ds.Where(d => d.StartsWith("begin")),
            b => ds.Where(d => d.StartsWith("end")))
        select String.Join(" ", list));

You can then get your results like so:

onelineStream.Merge(multiLineStream).Subscribe(d =>
{
    Console.WriteLine(d);
    Console.WriteLine();
});

This is what I got:

ok:michael
ok
begin:events 1:232 2:343 end:events
error:dfljsdf
error:fjkdjslf
ok

Let me know if that works for you.

Upvotes: 6

Related Questions