如何创建一个表示其他两个可观察对象完成的可观察对象？

Posted 2023-03-31

【中文标题】如何创建一个表示其他两个可观察对象完成的可观察对象？

【英文标题】：How to create an observable that signals the completion of two other observables?

【发布时间】：2021-12-14 17:03:34

【问题描述】：

我有两个IObservable<T>，并想创建一个IObservable<Unit>（或某种ISingle<Unit>，如果存在这样的东西），当两者都完成时发出完成信号，没有别的。在 Rx.NET 中执行此操作的惯用方式是什么？

我习惯了Reactor Core，在这种情况下我会使用flux1.then().and(flux2.then())。但我不相信 Rx.NET 提供 then 和 and 运算符。

【问题讨论】：

如果其中一个序列在另一个序列完成之前失败，那么理想的行为是什么？

理想情况下，一旦失败就应该抛出错误。

在这种情况下你可能想看看这个问题：How to merge two observables with early completion

谢谢@TheodorZoulias。

jackdry 实际上内置的 Merge 运算符具有您想要的行为。一旦其中一个序列失败，合并的序列也会立即失败。另一个序列，如果尚未完成，则取消订阅。

【参考方案1】：

假设两个序列属于同一类型，您可以使用Merge 运算符。将IgnoreElements 加入其中可能会有所帮助，以减少与可能包含大量元素的合并序列相关的同步压力。

IObservable<Unit> completion = sequence1.IgnoreElements()
    .Merge(sequence2.IgnoreElements())
    .Select(_ => Unit.Default);

如果序列的类型不同，您可以在合并之前将它们中的每一个投影到IObservable<Unit>，或者使用CombineLatest 而不是Merge。很多选择。

【讨论】：

谢谢！为什么我们需要LastOrDefaultAsync()？

@jackdry 好问题！我们不需要它，所以我将从答案中删除它。如果您想要Wait 或await 序列，则需要LastOrDefaultAsync，因为等待空序列会导致异常。