使用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方法将它们上载到服务器。问题是，当没有任何项目all chain onComplete事件正常运行，并且我的所有订阅者都获得此事件并继续它时，但当有一些项目并且所有项目都已上载到onComplete事件时，不会超过。doOnComplete（（）-

有人能帮忙找出这种行为的原因吗？

提前感谢！

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);
}