android - 带有 flatMapCompletable 的 RxJava 2 Observable 未完成

Question

我有一个 Observable 可以发出项目并将它们上传到服务器。

代码如下:

repository
            .getItems()
            .doOnComplete(() -> Log.d(TAG, "No items left."))
            .flatMapCompletable(item ->
                    repository
                            .uploadItems(item)
                            .onErrorComplete()
                            .andThen(
                                    deleteTemporaryItem()
                                            .onErrorComplete()
                            )
            );

getItems 方法逐个发出项目然后完成，uploadItems 方法将它们上传到服务器。问题是当没有项目时所有链 onComplete 事件都正常工作并且我的所有订阅者都收到此事件并继续它但是当有一些项目并且所有这些都被上传时 onComplete 事件不会比 .doOnComplete(() -> Log.d(TAG, "No items left.")) 方法，所有订阅者都不会收到此事件。我添加了 onErrorComplete 以确保 uploadItems 完成后的所有方法，我还在日志中看到所有方法都已完成，但来自 repository.getItems() 的 onComplete 事件并未发送给所有订阅者。

谁能帮忙找出造成这种行为的原因？

提前致谢!

Answer 1

请看这个例子:

我通过每个步骤传递项目，因此订阅者将收到有关已处理的每个项目的通知。处理管道涉及上传和删除文件。

请尝试更改实现并发布输出日志。

@Test
void name() throws Exception {
    Flowable<Integer> completed_work = Flowable.just(1, 2, 3)
            .map(integer -> integer * 1000)
            .flatMapSingle(integer ->
                    Completable.fromAction(() -> {
                        Thread.sleep(integer);
                        // do upload stuff here
                    })
                            .doOnComplete(() -> System.out.println("Uploaded file ...."))
                            //.timeout(10, TimeUnit.SECONDS)
                            .retry(3)
                            .andThen(
                                    Completable.fromAction(() -> {
                                        // do delete stuff...
                                    })
                                            .retry(2)
                                            //.timeout(10, TimeUnit.SECONDS)
                                            .doOnComplete(() -> System.out.println("Deleted file ..."))
                            )
                            .toSingle(() -> integer)
            )
            .doOnComplete(() -> System.out.println("Completed work"));


    completed_work.test()
            .await()
            .assertResult(1000, 2000, 3000);
}