c# - 将 IObservable 和 IObservable 合并到一个可观察量，当出现异常时，该可观察量会发生 OnErrors

Question

我正在尝试创建一个单一的 Observable，其中 OnNext 流来自一个 observable，而 OnError 流来自另一个 observable。

我这样做的原因是因为我试图包装一个不受我控制的类，并使用事件来传达其状态。它有两个事件，一个表示已完成(bool)，另一个表示发生异常。

IObservable<Exception> error = Observable.FromEventPattern<ExceptionRoutedEventArgs>(foo, "Failed")
               .Select(x => x.EventArgs.ErrorException);

IObservable<bool> opened = Observable.FromEventPattern<RoutedEventArgs>(foo, "Opened")
               .Select(x => ((Bar)x.Sender).IsOpen);

现在我无法使用标准 Observable.Merge 因为两个 observable 都有不同的通用参数。但在伪代码中我想完成这个:

Observable.Merge(opened, error, (op, err) =>
{
    if(op) { return op;}
    if(err != null){return Observable.Throw(err);}
}

现在有很多原因导致上面的代码与任何可能存在的代码完全不同，但我希望意图是明确的。

我认为实现这项工作的一种方法是使用主题<>，但我听说应该避免使用这些主题，因为它在功能概念中引入了状态。我的想法是，将两个可观察量组合成一个可观察量 OnNext 和 OnError 流似乎应该存在:)

Answer 1

您可以使用Observable.Create :

var combined = Observable.Create<bool>(o =>
{
    var openSub = opened.Subscribe(o);
    var errorSub = error.Subscribe(o.OnError);
    return new CompositeDisposable(openSub, errorSub);
});

或者您可以投影每个 Exception到IObservable<bool>只是抛出错误，并合并结果序列:

var combined = opened.Merge(error.SelectMany(Observable.Throw<bool>));

虽然仅仅因为您的源代码使用异常进行控制流并不意味着您必须这样做 - 为什么不直接投影 Exception至false ，那么你就得到 true成功并 false失败时:

var combined = opened.Merge(error.Select(_ => false));