如果第一个为空，则切换到另一个IObservable

Question

我正在编写一个函数，该函数检索有关主题的新闻并通过IObservable返回值反馈该新闻。

但是，我有几个新闻来源。我不想使用Merge将这些资源组合在一起。相反，我想做的是按优先级排序-

1. 调用我的函数时，将查询第一个新闻源（产生一个表示该源的IObservable）。
2. 如果该新闻源的IObservable完成但没有返回任何结果，则查询下一个新闻源。
3. 如果第二个来源完成但没有返回结果，则查询最终新闻来源。
4. 这整个行为被包装成一个可观察的状态，我可以返回给用户。

我可以使用内置的Rx扩展方法来完成这种行为吗？还是需要实现一个自定义类来处理此问题？我该怎么做？

Answer 1

在我看来，接受的答案是不可取的，因为它使用Subject, Do，并且当第一个序列不为空时仍然订阅第二个序列。如果第二个可观察量调用任何不平凡的东西，后者可能是一个大问题。我想出了以下解决方案：

public static IObservable<T> SwitchIfEmpty<T>(this IObservable<T> @this, IObservable<T> switchTo)
{
    if (@this == null) throw new ArgumentNullException(nameof(@this));
    if (switchTo == null) throw new ArgumentNullException(nameof(switchTo));
    return Observable.Create<T>(obs =>
    {
        var source = @this.Replay(1);
        var switched = source.Any().SelectMany(any => any ? Observable.Empty<T>() : switchTo);
        return new CompositeDisposable(source.Concat(switched).Subscribe(obs), source.Connect());
    });
}

该名称与现有的 RxJava 实现SwitchIfEmpty一致。这是关于将一​​些 RxJava 运算符合并到 RxNET 中的持续讨论。

我确信自定义IObservable实现会比我的更有效。您可以在这里找到由 ReactiveX 成员akarnokd编写的一篇文章。它也可以在NuGet上使用。