如何使用 Rx 从 SerialPort 读取的数据“重建行”

Question

我刚刚学习 Rx，并试图从使用 SerialPort 的 GPS 设备实现“NMEA 句子阅读器”。它是 GPS 数据这一事实对问题的重要性较低，所以让我们澄清一下 NMEA 格式由行组成，“$”符号表示新条目的开始，所以你得到的“句子”看起来类似于：

$[data for first line goes here]
$[data for second line goes here]
...

连接 SerialPort 的 DataReceived 事件非常简单：

 var port = new SerialPort("COM3", 4800);

 var serialPortSource = Observable.FromEventPattern<
    SerialDataReceivedEventHandler,
    SerialDataReceivedEventArgs>
    (
        handler => port.DataReceived += handler,
        handler => port.DataReceived -= handler
    ).Select(e => port.ReadExisting());

这给出了IObservable<string>，但显然这里返回的字符串不一定与 NMEA 句子对齐。例如，对于上面的示例数据，可以得到这样的序列：

$[data for first line g
oes here]\r\n$[data for
second line goes here]

如何正确地将其转换为一系列实际句子？在IEnumerable的世界里，我大概会从chars的序列开始，写一个类似这样的扩展方法：

public static IEnumerable<string> ToNmeaSentence(
    this IEnumerable<char> characters
)
{
    var sb = new StringBuilder();
    foreach (var ch in characters)
    {
        if (ch == '$' && sb.Length > 0) 
        {
            yield return sb.ToString();
            sb.Clear();
        }
        sb.Append(ch);
    }
}

现在我想知道在 Rx 中是否有这种操作的惯用方式？

Answer 1

它与 Enumerables 的代码完全相同。你使用Subscribe 代替快速枚举，你使用observer.OnNext 代替yield return。哦，你必须使用Observable.Create，因为 C# 没有像 Enumerables 那样对 Observers 提供语言支持（但是。这不是 Rx 的失败）。

Enumerables 和 Observables 完全一样。一个推，另一个拉。创建它们的语法略有不同。就是这样。

public static IObservable<string> ToNmeaSentence(
    this IObservable<char> characters
)
{
    return Observable.Create<string>(observer => {

        var sb = new StringBuilder();

        return characters.Subscribe(ch => {

            if (ch == '$' && sb.Length > 0)
            {
                observer.OnNext(sb.ToString());
                sb.Clear();
            }
            sb.Append(ch);

        });

    });
}

我通常不会在这么低的级别上进行编程，但是 Observables 并不比 Enumerables 更复杂。当人们第一次学习枚举时，很难理解。当人们第一次学习 Observables 时，很难理解。他们都做同样的事情，但一推一拉。除了这一区别之外，两者之间还有 1-1 的关系。

如果你认为 Rx 比 Enumerables 和 LINQ to Objects 更复杂，那就错了。当你还在学习它时，它只是看起来那样。

Answer 2

这是一个常见问题。我目前的想法是，您需要一个运算符来获取源序列并根据谓词将其分解为（连续的）窗口。

这个运算符可能很有用。

public static IObservable<IObservable<T>> WindowByExclusive<T>(this IObservable<T> input, Func<T, bool> isWindowBoundary)
{
    return Observable.Create<IObservable<T>>(o=>
    {
        var source = input.Publish().RefCount();
        var left = source.Where(isWindowBoundary).Select(_=>Unit.Default).StartWith(Unit.Default);
        return left.GroupJoin(
                        source.Where(c=>!isWindowBoundary(c)),
                        x=>source.Where(isWindowBoundary),
                        x=>Observable.Empty<Unit>(),
                        (_,window)=>window)
                    .Subscribe(o);
    });
}

它根据谓词有效地将源序列分解为切片。

您可以通过首先将字符串列表更改为chars 序列来使用它。 然后，您可以使用谓词c=='$' 应用WindowBy 运算符。 这将为您提供两个包含所需数据的窗口。 现在您已经分离了数据，然后您可以使用string.Join 将每个窗口中的字符连接成一个字符串。

完整的 LinqPad 示例

void Main()
{
    var data = new List<string>(){@"$[data for first line g", "oes here]\r\n$[data for ", "second line goes here]"};

    data.ToObservable()
        .SelectMany(s=>s)
        .WindowByExclusive(c => c=='$')
        .SelectMany(window=>window.ToList().Select(l=>string.Join(string.Empty, l)))
        .Where(s=>!string.IsNullOrEmpty(s))
        .Dump("WindowByExclusive");
}

// Define other methods and classes here
public static class ObEx
{
    public static IObservable<IObservable<T>> WindowByExclusive<T>(this IObservable<T> input, Func<T, bool> isWindowBoundary)
    {
        return Observable.Create<IObservable<T>>(o=>
        {
            var source = input.Publish().RefCount();
            var left = source.Where(isWindowBoundary).Select(_=>Unit.Default).StartWith(Unit.Default);
            return left.GroupJoin(
                            source.Where(c=>!isWindowBoundary(c)),
                            x=>source.Where(isWindowBoundary),
                            x=>Observable.Empty<Unit>(),
                            (_,window)=>window)
                        .Subscribe(o);
        });
    }
}

输出：

WindowByExclusive

[data for first line goes here]
[data for second line goes here]