refactoring - 任务 ToObservable 错误处理

Question

我有以下方法负责调用我的服务类并将结果传递给另一个方法以将它们保存在我的数据库中:

public IObservable<bool> SyncSessions()
{
    var subject = new ReplaySubject<bool>();

    try
    {
        var query = new ByFilterQuery { SearchPeriodStartTime = DateTime.Now };
        var sessions = _sessionService.GetSessions(query).Result;
        var saved = SaveSessions(sessions);
        subject.OnNext(saved);
        subject.OnCompleted();
    }
    catch (Exception ex)
    {
        subject.OnError(ex);
    }

    return subject;
 }

如果服务器返回 500 或类似的值，_sessionService.GetSessions 将抛出 HttpRequestException。我有一个模拟这种行为的单元测试，并希望测试我的方法是否优雅地处理错误。

有没有更好的方法以 Rx 方式传播错误？我试着做:

_sessionService.GetSessions(query).ToObservable().Select(SaveSessions);

但这引发了我的错误，而不是将其传递给调用方法中的错误处理操作。我还计划将此方法与其他几种方法合并，并在一个组合庄园中处理错误。

编辑:这是我订阅 observable 的方式

Exception error = null;
_sessionManager
    .SyncSessions()
    .Subscribe(null, e => error = e);

Assert.That(error, Is.Not.Null.After(500));

我将 null 传递给第一个参数，因为在此测试的上下文中我并不真正关心它

Answer 1

首先，对于您发布的原始代码，我建议使用 AsyncSubject而不是 ReplaySubject .将其视为针对单结果情况进行了优化。

现在回答你的问题...

每当可观察源产生异常时，该异常就会通过 OnError 传播。处理程序。如果，当您订阅 observable 时，您没有提供 OnError处理程序，然后将抛出异常。

您发布的小代码:

_sessionService.GetSessions(query).ToObservable().Select(SaveSessions);

是不足够的。没有同时发布显示您如何订阅 observable 的代码，我只能猜测您没有提供 OnError处理程序。

基于OP的测试代码编辑

您的模拟服务是否返回 Task抛出异常，还是模拟服务只是抛出异常？除非您使用 async/await，否则应该返回 Task 的函数，但在创建 Task 时抛出异常将立即引发异常而不是返回失败的 Task . ToObservable 甚至从未被调用，因为 GetSessions 抛出。如果您更改模拟服务以返回失败的 Task那么您的测试可能会正常工作。

如果你想捕捉即时异常以及失败的任务，那么你可以使用 Defer推迟执行您的方法，直到观察者订阅。除了失败的任务之外，它还将收集任何立即抛出的异常:

return Observable
    .Defer(() => _sessionService.GetSessions(query))
    .Select(SaveSessions);