Learning Rx: How can I parse an observable sequence of characters into an observable sequence of strings?

Question

This is probably really simple but I'm at the bottom of the learning curve with Rx. I've spent several hours reading articles, watching videos and writing code but I seem to have a mental block on something that seems like it should be really simple.

I'm gathering data from a serial port. I have used Observable.FromEventPattern to capture the SerialDataReceived event and convert it to an observable sequence of characters. So far so good.

Now, I want to parse that character sequence based on separator characters. There are no newlines involved, but each 'packet' of data is surrounded by a preamble and a terminator, both single characters. For the sake of argument, lets say they are braces { and }.

So if I get the character sequence j u n k { H e l l o } j u n k on my character sequence, then I want to emit either Hello or {Hello} on my string sequence.

I'm probably missing something simple but I can't even begin to figure out how to approach this. Any suggestions please?

Answer 1

This can be easily accomplished using Publish and Buffer:

var source = "junk{Hello}junk{World}junk".ToObservable();
var messages = source
    .Publish(o =>
    {
        return o.Buffer(
            o.Where(c => c == '{'),
            _ => o.Where(c => c == '}'));
    })
    .Select(buffer => new string(buffer.ToArray()));
messages.Subscribe(x => Console.WriteLine(x));
Console.ReadLine();

The output of this is:

{Hello}
{World}

The idea is that you can use the following opening and closing selectors in the call to Buffer. The use of Publish is to make sure that all three of Buffer, the opening selector, and the closing selector share the same subscription.

source:  junk{Hello}junk{World}junk|
opening: ----{----------{----------|
closing:     ------}|
closing:                ------}|