代码之家 › 专栏 › 技术社区 › Giuseppe Giacoppo

RxJava2-concat的最后一首成功单曲

rx-java2 reactive-programming observable java

Giuseppe Giacoppo · 技术社区 · 7 年前

我有N个单曲来源,我想收集这些,并获得最新的成功单曲。对于ex:

DB单个:

--[A] --|->

API单一:

---X--->

仅有一个的混凝土(DB、API):

--[A] --|->

否则:

DB单个:

--[A] --|->

API单一:

--[B] --|->

仅有一个的混凝土(DB、API):

--[B] --|->

这可能吗?我阅读了文档,但没有找到类似“lastsucessfullorerror()”的方法。我试过“elementAt”、“lastOrError”等,但他们的行为不是我想要的

非常感谢。

2 回复 | 直到 7 年前

akarnokd 7 年前

这可以通过以下方式实现:

public static <T> Single<T> latestSuccess(Single<T>... sources) {
     return Single.defer(() -> {
         AtomicReference<T> last = new AtomicReference<T>();
         return Observable.fromArray(sources)
             .concatMap(source ->
                  source.doOnSuccess(last::lazySet)
                  .toObservable()
                  .onErrorResumeNext(Observable.empty())
             )
             .ignoreElements()
             .andThen(Single.fromCallable(() -> {
                 if (last.get() == null) {
                     throw new NoSuchElementException();
                 }
                 return last.get();
             }));
     });
}

ricardopereira 4 年前

在我看来,一种很好的解决方案是使用 skip 操作员:

抑制由可观察到的

给出已通过的项目数 concat 减去1,让最后一个通过。

val listOfObservables = localChanges.map { myList ->
    pushChangeToServer(myList)
}

Observable.concat(listOfObservables)
    // Get only the latest:
    .skip(localChanges.size - 1)
    .map { lastResult ->
        return@map lastResult.data
    }

推荐文章