c# - 可观察管道中的异常处理

Question

我创建了一个 observable，它由一个项目通过运行异步方法转换为另一个项目组成。

IObservable<Summary> obs = scanner.Scans
                    .SelectMany(b => GetAssignment(b))
                    .SelectMany(b => VerifyAssignment(b))
                    .SelectMany(b => ConfirmAssignmentData(b))
                    .SelectMany(b => UploadAsset(b))
                    .Select(assignment => new Summary())
                    .Catch(LogException());

我想让这个防故障，所以如果在处理过程中抛出异常，我应该记录异常，但忽略异常并继续下一次扫描(由 scanner.Scans 推送的下一个项目)
当前代码捕获任何异常，但一旦抛出异常，序列就会结束。
我怎样才能让它“吞下”异常(记录它)，但要继续下一个项目？

Answer 1

这个问题假定了一个基本的误解:根据 Observable 契约，一个行为良好的 observable 在 OnError 之后终止。通知。对于您的情况，没有“只记录并继续”选项，因为没有什么可继续的。通过 OnError 抛出异常的 observable完成了，kaput，finito，一去不复返了。
评论提到 Retry ，这可能适用:如果您有一个像这样的可观察管道:

someHotSource
    .SelectMany(e => f(e)) //operator1
    .SelectMany(e => g(e)) //operator2
    .Subscribe(e => {});

然后异常可能发生在其中一个运算符(operator)中，杀死管道，但源可能仍然存在。 Retry然后将尝试重新创建具有相同功能的新管道。
您可以尝试使用 Materialize 来“欺骗”Observable 合约。和 Dematerialize ，但你会逆流而上。作弊的技巧是确保管道的任何部分都不会看到“原始” OnError ，因为该运算符将终止。而是 Materialize变成 OnError成 Notification ，它不会爆炸。那看起来像这样:
给定这样一个表现良好的管道:

var someHotSource = new Subject<int>();
var f = new Func<int, IObservable<int>>(i => Observable.Return(i));
var g = new Func<int, IObservable<int>>(i =>
{
    if(i % 13 == 0)
        return Observable.Throw<int>(new Exception());
    return Observable.Return(i);
});

var LogException = new Action<Exception>(e => Console.WriteLine("Exception"));
var p1 = someHotSource
    .SelectMany(e => f(e)) //operator1
    .SelectMany(e => g(e)) //operator2: suspect
    .Subscribe(e => Console.WriteLine(e));

...你可以这样作弊:

var p2 = someHotSource
    .SelectMany(e => f(e)) //operator1
    .SuspectSelectMany(e => g(e), LogException) //operator2: suspect
    .Subscribe(e => Console.WriteLine(e));

public static class X
{
    public static IObservable<Notification<T>> IgnoreOnCompleted<T>(this IObservable<Notification<T>> source)
    {
        return source
            .SelectMany(n => n.Kind == NotificationKind.OnCompleted
                ? Observable.Empty<Notification<T>>()
                : Observable.Return(n)
            );
    }
    
    public static IObservable<U> SuspectSelectMany<T, U>(this IObservable<T> source, Func<T, IObservable<U>> selector, Action<Exception> handler)
    {
        var x = source
            .Materialize()
            .SelectMany(e => selector(e.Value).Materialize().IgnoreOnCompleted()) //execute suspect selector, turn immediately into notifications
            .SelectMany(e =>
            {
                if (e.Kind == NotificationKind.OnError)
                {
                    handler(e.Exception);
                    return Observable.Empty<Notification<U>>();
                }
                else
                    return Observable.Return(e);
            }) //error logging/suppression
            .Dematerialize();
        return x;
    }
}

然后给出以下运行程序代码:

someHotSource.OnNext(1);
someHotSource.OnNext(12);
someHotSource.OnNext(13);
someHotSource.OnNext(15);

p1会炸。 p2将产生以下输出:

1
12
Exception
15