Rx:使用异步函数订阅并忽略错误 [英] Rx: subscribe with async function and ignore errors

问题描述

我想为 observable 中的每个项目调用一个异步函数.正如 此处 所回答，解决方案是使用 SelectMany.但是，如果异步方法抛出，订阅将终止.我有以下解决方案，这似乎有效:

I want to call an async function for each item in an observable. As answered here, the solution is to use SelectMany. However, if the async method throws, the subscription will terminate. I have the following solution, which seems to work:

obs.SelectMany(x => Observable
    .FromAsync(() => RunAsync())
    .Catch(Observable.Empty<string>()));

有没有更惯用的解决方案?

Is there a more idiomatic solution?

推荐答案

有一种标准方法可以观察 RunAsync 调用中发生的异常，那就是使用 .Materialize().

There is a standard way to be able to observe the exceptions that occur in your RunAsync call, and that's using .Materialize().

.Materialize() 方法将 IObservable 序列转换为 IObservable> 序列，您可以在其中可以对 OnNext、OnError 和 OnCompleted 调用进行推理.

The .Materialize() method turns an IObservable<T> sequence into a IObservable<Notification<T>> sequence where you can reason against the OnNext, OnError, and OnCompleted calls.

我写了这个查询:

var obs = Observable.Range(0, 10);

obs
    .SelectMany(x =>
        Observable
            .FromAsync(() => RunAsync())
            .Materialize())
    .Where(x => x.Kind != NotificationKind.OnCompleted)
    .Select(x => x.HasValue ? x.Value : (x.Exception.Message + "!"))
    .Subscribe(x => x.Dump());

使用此支持代码:

private int counter = 0;
private Random rnd = new Random();

private System.Threading.Tasks.Task<string> RunAsync()
{
    return System.Threading.Tasks.Task.Factory.StartNew(() =>
    {
        System.Threading.Interlocked.Increment(ref counter);
        if (rnd.NextDouble() < 0.3)
        {
            throw new Exception(counter.ToString());
        }
        return counter.ToString();
    });
}

当我运行它时，我得到这样的输出:

When I run it I get this kind of output:

2
4
5
1!
6
7
3!
10
8!
9

以 ! 结尾的每一行都是对导致异常的 RunAsync 的调用.

Each of the lines ending in ! are calls to RunAsync that resulted in an exception.

这篇关于Rx:使用异步函数订阅并忽略错误的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！