在 RxJava2 中正确使用 Single with Control flow

Question

我的代码中有以下Rx链：

public Single<List<Items>> dispatchStuff(){
    return mRepository.getList()
        .filter(list -> list.size() != 0)
        .flatMapSingle(mRepository::operateOnList())
        .doOnSuccess(mRepository::deleteList())
        .subscribeOn(mSchedulerProvider.io())
        .observeOn(mSchedulerProvider.ui());
}

但是，我不确定我在这里是否正确使用了过滤。getList()方法将始终返回值或错误，因此我认为使用Single是可以接受的。但是，仅当返回列表中有元素时，我才想执行其他操作。如果我过滤Single并且我的列表大小为 0，它将通过异常：

java.util.NoSuchElementException                                                                                                  
   at io.reactivex.internal.operators.maybe.MaybeFlatMapSingle$FlatMapMaybeObserver.onComplete(MaybeFlatMapSingle.java:106)                                                                                                  
   at io.reactivex.internal.operators.maybe.MaybeFilterSingle$FilterMaybeObserver.onSuccess(MaybeFilterSingle.java:92)                                                                                                  
   at io.reactivex.internal.operators.single.SingleFromCallable.subscribeActual(SingleFromCallable.java:37)                                                                                                  
   at io.reactivex.Single.subscribe(Single.java:2692)                                                                                                  
   at io.reactivex.internal.operators.maybe.MaybeFilterSingle.subscribeActual(MaybeFilterSingle.java:40)                                                                                                  
   at io.reactivex.Maybe.subscribe(Maybe.java:3707)                                                                                                  
   at io.reactivex.internal.operators.maybe.MaybeFlatMapSingle.subscribeActual(MaybeFlatMapSingle.java:47)                                                                                                  
   at io.reactivex.Single.subscribe(Single.java:2692)                                                                                                  
   at io.reactivex.internal.operators.single.SingleDoOnSuccess.subscribeActual(SingleDoOnSuccess.java:35)                                                                                                  
   at io.reactivex.Single.subscribe(Single.java:2692)                                                                                                  
   at io.reactivex.internal.operators.single.SingleSubscribeOn$SubscribeOnObserver.run(SingleSubscribeOn.java:89)                                                                                                  
   at io.reactivex.Scheduler$1.run(Scheduler.java:138)                                                                                                  
   at io.reactivex.internal.schedulers.ScheduledRunnable.run(ScheduledRunnable.java:59)                                                                                                  
   at io.reactivex.internal.schedulers.ScheduledRunnable.call(ScheduledRunnable.java:51)                                                                                                  
   at java.util.concurrent.FutureTask.run(FutureTask.java:237)                                                                                                  
   at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:269)                                                                                                  
   at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1113)                                                                                                  
   at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:588)                                                                                                  
   at java.lang.Thread.run(Thread.java:818)

当我使用 Single 时，我应该如何正确实现此控制流？

Answer 1

好吧，它不再是Single，而是Maybe.但是，如果您不想切换到 Maybe，也许这会有所帮助：

public Single<List<Items>> dispatchStuff(){
  return mRepository.getList()
    .flatMapSingle(list ->
        list.size() > 0
        ? mRepository.operateOnList(list)
          .doOnSuccess(mRepository::deleteList)
        : Single.just(list)
    )
    .subscribeOn(mSchedulerProvider.io())
    .observeOn(mSchedulerProvider.ui());
}

但是，也许您实际上应该有一个Flowable<Items>，而不是Single<List<Items>>？