IObservable 默认吞下异常？

Question

我有一个 IObservable 设置，我试图触发它从“冷”过渡到“热”。然而，纯粹靠运气，我发现在 IObservable 的某个地方，抛出了一个未处理的异常。

我立刻吓坏了，因为 Reactive Extensions 没有让我知道这个未处理的异常。 我猜这个流刚刚终止，而不是。现在我担心 Rx Extensions 会吞噬其他异常。

这是我目前用来触发 IObservable 为“热”的代码。我希望 IObservable 内部发生的任何未处理的异常都会冒泡并被抛出这里。但他们不是。

var observable = Observable.Create<>(async a =>
  { 
    ... 
    a.OnNext();
    ...
    a.OnCompleted();
  });

observable = observable.Do(onNext: ..., 
  onCompleted: async () =>
  {
    // This throws the unhandled exception
    await MethodThatThrowsExceptionAsync();
  });


// I would expect any exceptions inside the IObservable to bubble up and be rethrown here.
await observable.LastOrDefaultAsync();  

我做错了吗？这是预期的行为吗？如果是这样，这似乎极易出错。

Answer 1

Do 用于表达通知的副作用，而不是实际修改通知本身。一般不建议在Do 中抛出错误。

如果我们把它分解，你想在 observable 完成时运行一个任务，你可能会抛出一个异常，它会冒泡。

有一个操作员可以在完成后分流另一个 observable - Concat。

    var observable = Observable.Create<int>(async a =>
    {
        await Task.Delay(1000);
        a.OnNext(0);
        a.OnCompleted();
    });

    observable = observable.Concat(Observable.FromAsync(async () =>
      {
          await Task.Delay(1000); //simulate work

          // This throws the unhandled exception
          throw new Exception("I'm from async");              

          return 1; //type inference
      }));


    // This now throws
    await observable.LastOrDefaultAsync();

附言

使用async/await 代替Action<> 时要小心。 他们默默地失败了。

异步无效方法considered harmful。