c# - 如何级联具有不同错误处理的可观察对象

Question

我有一个简单的 IObservable，它提供来自串行设备的 NMEA 字符串:

     var source = Observable.Create<string>(
            observer =>
            {
                var port = new SerialPort("COM4", 4800, Parity.None, 8, StopBits.One);

                string output = string.Empty;

                port.DataReceived += (sender, eventArgs) =>
                {
                    var input = (SerialPort)sender;
                    var buffer = new byte[input.BytesToRead];

                    try
                    {
                        input.Read(buffer, 0, buffer.Length);
                    }
                    catch (Exception exception)
                    {
                        observer.OnError(exception);
                    }

                    var encoding = Encoding.ASCII;
                    var data = encoding.GetString(buffer);

                    if (data.StartsWith("$") && output != string.Empty)
                    {
                        if (output.StartsWith("$") || data.EndsWith(Environment.NewLine))
                        {
                            output = output.TrimEnd(Environment.NewLine.ToCharArray());
                            observer.OnNext(output);
                        }

                        output = data;
                    }
                    else if ((output == string.Empty || data.StartsWith("$")) || output.StartsWith("$"))
                    {
                        output += data;
                    }
                };

                try
                {
                    port.Open();
                }
                catch (Exception error)
                {
                    observer.OnError(error);
                }

                return port;
            }
        );

现在我想将这些字符串解析为具体的消息类型，以便稍后进行过滤。因此，IObservable 需要像 IObservable 一样更专业。

这可以通过将结果重新投影到 NMEAMessage 中的 .Select() 语句来解决...(以下只是一个简单的重投影的例子)

var selected = source.Select(s => { return new NmeaMessage(s); });

...但是如果无法进行重新投影会发生什么(例如，未知的消息类型或传递的字符串的解析错误)。如何处理？我不能在这里调用 OnError(因为它是一个 Observable 而不是一个 Observer。在这种情况下只是抑制解析错误并且不返回任何内容？如何声明源可能不是有效的 NMEA 源？我应该创建一个虚拟“设备” "类(使用内部字符串源)而不是级联或过滤 Observables？“Device”类可以使用事件，然后可以再次创建顶部的 Observable(Observable.FromEventPattern<>)。

我还希望“观察”解析器能够订阅不同来源的 IObservable。将解析器和重投影集成到此场景中的最佳方法是什么？

Answer 1

包装你的 NmeaMessage像这样的静态函数中的构造函数:

public static NmeaMessage TryParseNmeaMessage(TInputData d)
{
    if (IsValidInput(d))
        return new NmeaMessage(d);
    else
        return null;
}

然后你可以这样做:

var inputData = Observable.Create(...);
var parsed    = inputData.Select(d => TryParseNmeaMessage(d))
                         .Where(d => d != null);

显然你需要定义 IsValidInput()也是。

您还可以从 ParseMessage 返回一些 BadNmeaMessage(包含错误信息的 NmeaMessage 的子类)而不是 null。
然后您可以使用 .OfType<(Bad)NmeaMessage>() 分别对好消息/坏消息使用react.

或者您可以发出 OnError 信号(在 ParseMessage 中抛出)，然后重新启动序列。

参见 here用于高级错误处理。

有一篇关于 F# 中这种错误处理的优秀文章:Railway oriented programming .
相同的原则可以在 C# 中与 RX 一起使用。