如何使用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#没有对Observers有语言支持,就像它对Enumerables一样(但是.这不是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更复杂.当人们第一次学习Enumerables时,很难理解.当人们第一次学习Observables时,很难理解.他们两个都做同样的事情,但一个推,一个拉.除了那一个区别之外,两者之间存在1-1的关系.

你认为Rx比Enumerables和LINQ to Objects更复杂是错误的.当你还在学习它时,它就会出现.