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

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。

这可以通过 .Select() 语句解决，其中 Result 被重新投影到 NMEAMessage... （以下只是一个简单的重投影示例）

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

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

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

有关高级错误处理，请参阅 here。

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