c# - 是否有一个 RX 扩展来从另一个只捕获异常的 observable 创建一个 observable

Question

功能上与此等效的东西

//someObservable is an observable from another part of my application which produces Command objects
//can also have a ConnectionException

IObservable<Command> someObservable;

Subject<Command> sub = new Subject<Command>();

someObservable.Subscribe(c => { }, x =>
{
        sub.OnError(x);
});

这将允许我对异常有一个可观察的，以便我可以在其他地方处理它们

Answer 1

您可以使用 Materialize/Dematerialize 过滤除异常之外的所有内容。例如：

var subject = new Subject<int>();

var onlyExceptions = subject.Materialize().Where(n => n.Exception != null).Dematerialize();

subject.Subscribe(i => Console.WriteLine("Subscriber 1: {0}", i),
                    ex => Console.WriteLine("Subscriber 1 exception: {0}", ex.Message));

onlyExceptions.Subscribe(i => Console.WriteLine("Subscriber 2: {0}", i),
                            ex => Console.WriteLine("Subscriber 2 exception: {0}", ex.Message));

subject.OnNext(123);
subject.OnError(new Exception("Test Exception"));

Answer 2

此扩展方法只会返回单个 OnError，并且当且仅当源发送 OnError 时。

public static IObservable<T> ErrorsOnly<T>(this IObservable<T> source)
{
    return source.IgnoreElements().Concat(Observable.Never<T>());
}

虽然不确定它在实践中有多有用。如果您不关心 OnCompleted(); 的话， IgnoreElements() 可能就足够了。

Answer 3

不要忘记，一旦出现可观察到的错误，它就结束了。您当然可以订阅，或使用 Observable.Catch 或 finally 扩展方法之一，但您只会得到一个错误。

Answer 4

您可以有一个只处理错误的订阅者（并且对序列中的常规项目不执行任何操作），并使用 Catch() 来消除其他任何地方的错误。