如何合并两个可观察的早期完成

Posted 2023-03-31

【中文标题】如何合并两个可观察的早期完成

【英文标题】：How to merge two observables with early completion

【发布时间】：2020-04-03 06:33:15

【问题描述】：

内置Merge 运算符的行为是在两个源都完成时完成。我正在寻找这个运算符的变体，它产生一个可观察的，当两个源可观察的 any 完成时完成。例如，如果第一个 observable 成功完成，稍后第二个 observable 完成但出现异常，我希望忽略此异常。

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

/// <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.

【参考方案1】：

有趣的小技巧：

public static IObservable<T> MergeUntilAnyCompletes<T>(this IObservable<T> first,
    IObservable<T> second)

    return Observable.Merge(
        first.Materialize(),
        second.Materialize()
    ).Dematerialize();

Materialize 将 observable 转换为通知的 observable，因此 Merge 将不再抑制 OnCompleted 通知。当您Dematerialize 时，该操作员将看到OnCompleted 并停止。

旁注：如果您想要一些关于Materialize/Dematerialize 的有趣、略带学术性的阅读，请阅读this blog post。他在写关于 Ix 的文章，但同样的事情也适用于 Rx。

【讨论】：

这是一个不错的解决方案！

blog post 也很有启发性！虽然我在System.Interactive API 中找不到 Materialize/Dematerialize 运算符。

有趣。这篇文章距 Rx 和 Ix 的一位原始开发者已有 10 年的历史，所以事情发生了变化也就不足为奇了。但令人惊讶的是，它被删除而没有明显的等价物。

2021 年更新：const shortMerge = (...sourceObservables) => merge(...sourceObservables.map((sourceObservable) => sourceObservable.pipe(materialize()))).pipe(dematerialize())