学习Rx:如何将一个可观察的字符序列解析为一个可观察的字符串序列

Question

这可能真的很简单，但我正处于Rx学习曲线的底部。我花了几个小时阅读文章，看视频和写代码，但我似乎在一些看起来应该非常简单的事情上有一个心理障碍。

我正在从串行端口收集数据。我使用Observable.FromEventPattern捕获SerialDataReceived事件并将其转换为可观察的字符序列。到目前为止一切顺利。

现在，我要基于分隔符解析该字符序列。没有换行符，但是每个"数据包"都有一个序言和一个结束符，都是单个字符。为了便于讨论，我们设它们是大括号{和}。

如果我得到字符序列j u n k { H e l l o } j u n k我的字符序列，然后我想在字符串序列上发出Hello或{Hello}。

我可能错过了一些简单的东西，但我甚至不知道如何处理这个问题。有什么建议吗?

Answer 1

这可以使用Publish和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();

它的输出是:

{Hello}
{World}

的想法是，您可以在调用Buffer中使用以下打开和关闭选择器。使用Publish是为了确保所有三个Buffer，打开选择器和关闭选择器共享相同的订阅。

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

Answer 2

使用Scan将到目前为止接收到的值聚合到聚合字符串中(TAccumulate是string)，并在每次获得结束大括号时将该字符串重置为""。(我将把实现聚合函数的工作留给您。)这将产生像

这样的可观察对象

j
ju
jun
junk
junk{
junk{h
junk{hi
junk{hi}
j
ju
...

然后你可以使用Where只发出以}

结尾的字符

然后最后使用Select去掉junk .

应该写成

IObservable<string> packetReceived = 
  serialPort.CharReceived
    .Scan(YourAggregationFunction)
    .Where(s => s.EndsWith("}"))
    .Select(s => s.EverythingAfter("{"));

(我把EverythingAfter留给你来实现)。

请注意，当您在试验聚合函数时，使用string的IEnumerable接口来测试它可能更容易，即

foreach (s in "junk{hi}hunk{ji}blah".Scan(YourAggregationFunction))
  Console.WriteLine(s);

好的，这里有一个完整的工作示例

static void Main(string[] args) {
    var stuff = "junk{hi}junk{world}junk".ToObservable()
        .Scan("", (agg, c) => agg.EndsWith("}") ? c.ToString() : agg + c)
        .Where(s => s.EndsWith("}"))
        .Select(s => s.Substring(s.IndexOf('{')));
    foreach (var thing in stuff.ToEnumerable()) {
        Console.WriteLine(thing);
    }
}