如何使用Reactive Extensions解析串口的字符流？

Question

我需要解析来自测试仪器的串行数据流,这似乎是Reactive Extensions的一个很好的应用程序.

协议非常简单......每个"数据包"都是一个字母后跟数字.对于每种数据包类型,数字位数是固定的,但可以在数据包类型之间变化.例如

... ... A1234B123456C12

我试图把它分解为一个Observable of Strings,例如

"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转换为String的Observable,其中每个字符串都是一个数据包？

Answer 1

这是一个略短的方法,与James的第一个解决方案相同,具有类似的辅助方法:

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);

该Scan部分构建了以字母结尾的字符串,例如:

A
A1
A12
A123
...

在Where那么就选择那些正确的长度.基本上它只是从James'中删除了Tuple并使用了字符串长度.