如何阻止 IConnectableObservable.Wait() 吞下未处理的异常？

Question

我有一个使用 Rx.NET 的控制台应用程序。

我需要阻塞直到 IConnectableObservable 完成，为此我使用IConnectableObservable.Wait()。

当一个未处理的异常被抛出时，它被吞下并且应用程序挂起。我希望应用程序崩溃，并将堆栈跟踪打印到控制台。

我不想将OnError 处理程序添加到我的IConnectableObserver，因为这会丢失原始堆栈跟踪。

我曾尝试在未发布的 observable 上使用 .Wait() 方法，但这会重新订阅会导致不良行为。

我尝试过使用.GetAwaiter().GetResult()，但这有同样的问题。

var connectable = myObservable.Publish();

connectable.Subscribe(myObserver1);
connectable.Subscribe(myObserver2);

connectable.Connect();
connectcable.Wait();

如何在保留典型的未处理异常行为的同时等待 IConnectableObservable 完成？

Answer 1

这里的事件链中有一些误导。 错误并没有被吞没 - 远非如此，它正在被重新抛出。

通常怀疑是一些奇怪的并发和调度问题，但没有人怀疑Subscribe 方法。

当您使用自己的IObserver<T> 以外的其他内容调用Subscribe 时，您正在使用这些默认操作创建AnonymousObserver。

new AnonymousObserver<T>(Ignore, Throw, Nop)

这是有效的

new AnonymousObserver<T>(_ => {}, exn => throw exn, () => {})

默认错误处理程序将在您观察的任何上下文中抛出错误。哎呀。有时它可能是 AppDomain 计时器，或者在一个池线程上，由于无法处理，您的应用程序出现故障。

因此，如果我们将示例更改为在虚拟处理程序中提供，

var myObservable = Observable.Interval(TimeSpan.FromMilliseconds(100)).Take(4).Concat(Observable.Throw(new Exception(), 1L));
var connectable = myObservable.Publish();

connectable.Subscribe(Console.WriteLine, exn => Console.WriteLine("Handled"));
connectable.Subscribe(Console.WriteLine, exn => Console.WriteLine("Handled"));

connectable.Connect();

try
{
    connectable.Wait();
}
catch (Exception)
{
    Console.WriteLine("An error, but I'm safe");
}

您可以按预期处理Wait 中的错误。