学习 Rx：如何将可观察的字符序列解析为可观察的字符串序列？

Question

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

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

现在，我想根据分隔符解析该字符序列。不涉及换行符，但每个 'packet' 数据都被一个前导码和一个终止符包围，都是单个字符。为了争论，假设它们是大括号 { 和 }.

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

我可能遗漏了一些简单的东西，但我什至无法开始弄清楚如何解决这个问题。有什么建议吗？

Answer 1

使用 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);
    }
}

Answer 2

这可以使用 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:                ------}|

学习 Rx：如何将可观察的字符序列解析为可观察的字符串序列？

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

c#

system.reactive