java - RxJava2习惯用法，用于合并两个Maybes

给定两个Maybe值，我如何将它们组合为一个Maybe，该代码可以:

每当源onSuccess()中的任何一个调用Maybes

时都调用onSuccess

每当源onComplete()都调用Maybes时调用onComplete()？

(参见Scala或Vavr中的Option.orElse()。)

例如，假设存在执行我想要的操作的combine()方法:

combine(Maybe.just(a), Maybe.empty())  ≍ Maybe.just(a)
combine(Maybe.empty(), Maybe.just(b))  ≍ Maybe.just(b)
combine(Maybe.empty(), Maybe.empty())  ≍ Maybe.empty()
combine(Maybe.never(), /*anything*/ )  ≍ /*the thing*/
combine(/*anything*/,  Maybe.never())  ≍ /*the thing*/

起初，我以为amb()和family是我想要的，但是只要其中一个源Maybe完成，它就完成了，这意味着如果第一个Maybe完成时没有值，则您永远不会从第二个Maybe中获取值。

现在我正在使用

Maybe.mergeArray(m1, m2).firstElement()

这似乎可以满足我的要求，但我不确定它是否正确，也不确定这是最干净的方法。 (例如，如果有一些延迟，它会在一个或另一个来源同时调用onSuccess()还是会等待两个onComplete()？)

这样对吗？有没有更惯用的方法？

预计到达时间:我很高兴取得第一个价值；我不需要等待两者都完成:

combine(Maybe.just(a), Maybe.just(b))  ≍ Maybe.just(/* a or b, don't care */)

(我可以想象在某些情况下我可能更喜欢一个或另一个，并希望通过参数的顺序来表明这一点，但是在那种情况下，我怀疑顺序会比并行好。)

最佳答案

有一种稍微不同的方法，可能与您的定义更接近。这将使用 Observable.switchMapMaybe() :

将上游项映射为MaybeSources并切换(订阅)到较新的项，同时处置较旧的项(并忽略它们的信号)，并发出当前项的最新成功值(如果可用)，如果此Observable或任何 Activity 的内部组件立即失败MaybeSources失败。

Observable.just(m1, m2).switchMapMaybe(m -> m).firstElement()

但是使用Maybe.mergeArray(m1, m2).firstElement()的方法也应该足够。 firstElement()运算符发出mergeArray() flowable发出的第一个元素。这是无序的，因此没有任何可能完成的信息。