Что может заставить BehaviorSubject сообщать о необработанных исключениях другого несвязанного наблюдаемого? ⇐ C#

[Anonymous]

Контекст
Я исследую сложную проблему, когда по какой-то причине BehaviorSubject выдает наблюдателям ошибку. Это происходит даже несмотря на то, что на самом деле никто не взаимодействует с этой наблюдаемой. Единственная связь, по-видимому, заключается в том, что они оба оцениваются в потоке пользовательского интерфейса моего приложения WPF.

Это в некоторой степени отражает то, как выглядит мой код, но, скорее всего, в нем отсутствует реальная проблема:

частный BehaviorSubject, доступный только для чтения _exampleSubject; общественная недействительность NotifyChanged (int newValue) { _exampleSubject.OnNext(newValue); } общественный IObservable _ConstructStream() { вернуть _exampleSubject .Делать( onNext: count => Console.WriteLine($"_exampleSubject.OnNext: {count} (поток: {Thread.CurrentThread.ManagedThreadId})"), onError: исключение => Console.WriteLine($"_exampleSubject.OnError: {исключение} (поток: {Thread.CurrentThread.ManagedThreadId})"), onCompleted: () => Console.WriteLine($"_exampleSubject.OnCompleted (thread: {Thread.CurrentThread.ManagedThreadId})")); } По какой-то причине _exampleSubject выдает исключение, что не имеет смысла, поскольку OnError никогда и нигде не вызывается. В другом месте исключение не обрабатывается должным образом в наблюдаемом объекте, что, похоже, каким-то образом мешает ему.

К сожалению, мне не удалось создать минимальный пример, воспроизводящий эту проблему. Кажется, существует какое-то состояние гонки, которое возникает только в том случае, если включен весь код.

Мне удалось воспроизвести часть проблемы, описанной ниже.
Настройка
Чтобы воспроизвести проблему, используйте шаблон WPF по умолчанию и внесите следующие изменения в App.xaml.cs:

// Целевая платформа: // net7.0-окна // Пакеты: // Система.Реактивная 6.0.0 пространство имен WpfApp1 { Общедоступное частичное приложение класса: Приложение { защищенное переопределение void OnStartup (StartupEventArgs e) { base.OnStartup(e); // ... Вызовите здесь методы примера. } // ... Определите здесь примеры методов. } } Пример 2 частная асинхронная задача _Example_2() { вар диспетчерSynchronizationContext = новый DispatcherSynchronizationContext(Application.Current.Dispatcher); // Добавление этого кода больше не будет воспроизводить проблему: // используем var disposable = Observable var одноразовый = наблюдаемый .Timer(TimeSpan.FromMilli Seconds(100)) .ObserveOn(dispatcherSynchronizationContext) .SubscribeOn(dispatcherSynchronizationContext) .Subscribe(_ => создать новое исключение InvalidOperationException()); // Подождите немного, пока что-то пойдет не так. await Task.Delay(TimeSpan.FromMilli Seconds(300)); } Когда генерируется исключение, которое, по-видимому, замораживает основной поток, все окно зависает, и все остальное в том же планировщике перестает обрабатываться. Другие наблюдаемые, которые не привязаны к тому же планировщику, продолжаются некоторое время, прежде чем приложение автоматически закроется.

Это то, чего я и ожидал, поскольку необработанное исключение должно завершить работу приложения. К сожалению, нигде нет сообщений журнала, указывающих на то, что это произошло.

Проблема 1: Странно то, что если я добавлю оператор using в одноразовый файл, проблема больше не возникнет. Я ожидал, что удаление произойдет в конце функции, после ожидания задержки, но, похоже, это происходит до этого.

Глядя на это еще раз, я запутался в собственном примере. Через 300 мс он вернется из метода, который выполняет вызов Dispose. В голове я думал об этом больше как о Timeout.InfiniteTimeSpan.
Пример 1
Интересно, что ту же проблему можно воспроизвести с помощью Observable.FromAsync с помощью Concat, но только если перед переключением не происходит задержки.

частная асинхронная задача _Example_1() { вар диспетчерSynchronizationContext = новый DispatcherSynchronizationContext(Application.Current.Dispatcher); наблюдаемый .Timer(TimeSpan.FromMilli Seconds(100)) .ObserveOn(dispatcherSynchronizationContext) .SubscribeOn(dispatcherSynchronizationContext) .Select(index => Observable.FromAsync(async() => { // Добавление этого кода больше не будет воспроизводить проблему: // ждём Task.Delay(TimeSpan.FromMilli Seconds(100)); выбросить новое InvalidOperationException(); })) .Конкат() .Подписаться(); // Подождите немного, пока что-то пойдет не так. await Task.Delay(TimeSpan.FromMilli Seconds(300)); } Проблема 2: Кажется, происходит что-то странное, когда поток переключается и выполнение приостанавливается. Кажется, это меняет поведение. Добавление задержки все равно должно привести к сбою диспетчера.