默认情况下可观察到的吞咽异常？

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中抛出错误。

如果我们将其分解，您希望在可观察对象完成时运行一个任务，在该任务中您可能会引发一个异常，该异常会冒泡。

[有一个操作员可以在完成后分流另一个可观察的对象-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();

P.S。

[使用async/await代替Action<>时要小心。

异步无效方法considered harmful。