如何使用反应式扩展来解析来自串口的字符流？

我需要解析来自测试仪器的串行数据流，这似乎是反应式扩展的一个优秀应用程序。

该协议非常简单......每个“数据包”都是一个字母后跟数字。每种数据包类型的数字位数是固定的，但不同数据包类型的数字位数可能不同。例如

...A1234B123456C12...

我试图将其分解为一个可观察的字符串，例如

“A1234”“B123456”“C12”...

认为这很简单，但没有看到明显的方法来解决这个问题（我有一些 LINQ 的经验，但对 Rx 来说是新手）。

这是我到目前为止的代码，它从串行端口的 SerialDataReceived 事件生成可观察的字符。

        var serialData = Observable
                            .FromEventPattern<SerialDataReceivedEventArgs>(SerialPortMain, "DataReceived")
                            .SelectMany(_ => 
                            {
                                int dataLength = SerialPortMain.BytesToRead;
                                byte[] data = new byte[dataLength];
                                int nbrDataRead = SerialPortMain.Read(data, 0, dataLength);

                                if (nbrDataRead == 0)
                                    return  new char[0]; 

                                 var chars = Encoding.ASCII.GetChars(data);

                                 return chars; 
                            });

如何将serialData转换为Observable of String，其中每个字符串都是一个数据包？

这是一个稍微短一点的方法，与詹姆斯的第一个解决方案 https://stackoverflow.com/a/24396881/337065，使用类似的辅助方法：

public static bool IsCompletePacket(string s)
{
    switch (s[0])
    {
        case 'A':
            return s.Length == 5;
        case 'B':
            return s.Length == 6;
        case 'C':
            return s.Length == 7;
        default:
            throw new ArgumentException("Packet must begin with a letter");
    }
}

那么代码是：

var packets = chars
    .Scan(string.Empty, (prev, cur) => char.IsLetter(cur) ? cur.ToString() : prev + cur)
    .Where(IsCompletePacket);

The Scan部分构建以字母结尾的字符串，例如：

A
A1
A12
A123
...

The Where然后只选择那些长度正确的。本质上，它只是从 James 中删除元组并使用字符串长度代替。