如何将两个可观测数据与早期完成合并

Question

内置Merge操作符的行为是在和源都完成时完成的。我正在寻找这个操作符的一个变体，它产生一个可观察的，当完成两个源可观测值的任何时完成的可观察性。例如，如果第一个可观测的完成成功，然后通过一个异常完成第二个可观测的完成，我希望忽略这个异常。

我想出了一个实现，将一个特殊的哨兵异常连接到两个可枚举项中，然后合并的序列捕获并抑制这个异常。我想知道我是不是错过了一个更简单的解决方案。

/// <summary>
/// Merges elements from two observable sequences into a single observable sequence,
/// that completes as soon as any of the source observable sequences completes.
/// </summary>
public static IObservable<T> MergeUntilAnyCompletes<T>(this IObservable<T> first,
    IObservable<T> second)
{
    var sentinel = new Exception();
    first = first.Concat(Observable.Throw<T>(sentinel));
    second = second.Concat(Observable.Throw<T>(sentinel));
    // Concat: Concatenates the second observable sequence to the first
    // observable sequence upon successful termination of the first.
    return first.Merge(second)
        .Catch(handler: (Exception ex) =>
        ex == sentinel ? Observable.Empty<T>() : Observable.Throw<T>(ex));
    // Catch: Continues an observable sequence that is terminated by an exception
    // of the specified type with the observable sequence produced by the handler.
}

Answer 1

有趣的黑客：

public static IObservable<T> MergeUntilAnyCompletes<T>(this IObservable<T> first,
    IObservable<T> second)
{
    return Observable.Merge(
        first.Materialize(),
        second.Materialize()
    ).Dematerialize();
}

Materialize将可观察到的通知转换为可观察的通知，因此Merge将不再禁止OnCompleted通知。当您Dematerialize时，该操作符将看到OnCompleted并停止。

附带注意:如果你想要一些有趣的，略带学术性的关于Materialize/Dematerialize的阅读，读这篇博客文章。他写的是Ix，但同样的事情也适用于Rx。