平面图不继续Rx链，但平面图与完成与然后（Observable.just（真）工作？

Question

我试图将一个完整的链接到我的Rx链中，当我这样做时，链永远不会在onError或onComplete中完成。

当我单步执行代码时，我的可完成代码就会被执行。我甚至可以添加日志记录并看到它登录到它自己的doOn完成（）

下面将记录“我已完成”，但不会进入错误或完成回调。

 profileRepo.getLocalProfileIfAvailableElseRemote()
                .flatMapCompletable { profile ->
                    userRoutingRepo.disableRule(profile.account_uid, userRoutingRule.id)
                            .doOnComplete {
                                Log.i("I COMPLETED", "I COMPLETED")
                            }
                }
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribeBy(
                        onError = { error ->
                            //do error
                        },
                        onComplete = {
                           //do success
                        }
                ).addTo(disposable)

如果我改为使用flatMap并使用andThen返回布尔可观测值，它将起作用

 profileRepo.getLocalProfileIfAvailableElseRemote()
                .flatMap { profile ->
                    userRoutingRepo.disableRule(profile.account_uid, userRoutingRule.id)
                            .doOnComplete {
                                Log.i("I COMPLETED", "I COMPLETED")
                            }.andThen(Observable.just(true))
                }
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribeBy(
                        onError = { error ->
                         //do error
                        },
                        onNext = {
                           //do next
                        }
                ).addTo(disposable)

我尝试在flatMapCompletable版本中添加“and then”并调用Completable。complete（）但这也不行吗？

我不明白为什么completable正在完成，但拒绝使用flatMapCompletable？

编辑：这是我的完整尝试的更新，但不起作用

注意用户路由服务。disableRule（accountUid，ruleId）是改装接口

 profileRepo.getLocalProfileIfAvailableElseRemote()
                .flatMapCompletable { profile ->
                    userRoutingRepo.disableRule(profile.account_uid, userRoutingRule.id)
                            .andThen(Completable.complete())
                }
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribeBy(
                        onError = { error ->
                          Log.i("TAG", "ERROR")
                        },
                        onComplete = {
                            Log.i("TAG", "COMPLETE")
                        }
                ).addTo(disposable)

 override fun disableRule(accountUid: String, ruleId: String): Completable {
        return activeStateToggler(userRoutingSourceApi.disableRule(accountUid, ruleId),
                ruleId,
                false)
    }



override fun disableRule(accountUid: String, ruleId: String): Completable {
        return userRoutingService.disableRule(accountUid, ruleId)
                .doOnError { error ->
                    authenticationValidator.handleAuthenticationExceptions(error)
                }
    }

    private fun activeStateToggler(completable: Completable,
                                   ruleId: String,
                                   stateOnSuccess: Boolean
    ): Completable {
        return completable
                .doOnSubscribe {
                    stateTogglingInProgress.add(ruleId)
                }
                .doOnComplete {
                    stateTogglingInProgress.remove(ruleId)
                    getLocalUserRule(ruleId)?.active = stateOnSuccess
                    stateTogglingInProgressPublishSubject.onNext(UserRoutingStateToggleSubjectType.Success)
                }
                .doOnError {
                    stateTogglingInProgress.remove(ruleId)
                    stateTogglingInProgressPublishSubject.onNext(UserRoutingStateToggleSubjectType.Error(
                            it))
                }
    }

Answer 1

使用flatMapCompletable时，需要返回Completable。自己完成（）。

编辑：

 profileRepo.getLocalProfileIfAvailableElseRemote()
     .flatMap { profile ->
         userRoutingRepo.disableRule(profile.account_uid, userRoutingRule.id)
             .doOnComplete { Log.i("I COMPLETED", "I COMPLETED") } }
     .flatMapCompletable { () -> { Completable.complete() } }
     .subscribeOn(Schedulers.io())
     .observeOn(AndroidSchedulers.mainThread())
     .subscribeBy(
         onError = { error ->
             //do error
         },
         onNext = {
             //do next
         }
    ).addTo(disposable)

编辑2：因为一次性规则是一个完整的

 profileRepo.getLocalProfileIfAvailableElseRemote()
     .flatMapCompletable { profile ->
         userRoutingRepo.disableRule(profile.account_uid, userRoutingRule.id)
             .doOnComplete { Log.i("I COMPLETED", "I COMPLETED") }
             .andThen(Completable.complete().doOnCompleted { Log.i("comp2", "comp2")) }
     .subscribeOn(Schedulers.io())
     .observeOn(AndroidSchedulers.mainThread())
     .subscribeBy(
         onError = { error ->
             //do error
         },
         onNext = {
             //do next
         }
    ).addTo(disposable)

编辑3：工作样本

Observable.just(1)
    .flatMapCompletable { profile ->
        Completable.complete()
            .doOnComplete { Log.i("I COMPLETED", "I COMPLETED") }
            .andThen(Completable.complete().doOnComplete { Log.i("I COMPLETED", "I COMPLETED 2") })}
    .subscribeBy(
        onError = { error ->
        },
        onComplete = {
            Log.d("I COMPLETED", "I COMPLETED 3")
        })

Answer 2

这就是flatMapCompletable所做的：

将上游可观察的每个元素映射到CompletableSources中，订阅它们并等待上游和所有CompletableSources完成。

使用flatMapCompletable时，返回的Completable将等待上游的可观察终端事件（onComplete）。

使用flatMapCompletable时，仅当您确定链中的所有内容都已完成时才使用它。

在您的情况下，它不起作用，因为您的源Observable是热的，并且永远不会完成。