FlatMapCompletable 不会继续 Rx 链，但是带有 "andThen(Observable.just(true)" 的可完成的平面图可以工作吗？

Question

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

当我单步执行代码时，我的可完成代码被执行。我什至可以添加日志记录并看到它登录到它自己的 doOnComplete()

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

 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 版本中添加“andThen”并调用 Completable.complete() 但这也不起作用？

我不知道为什么我的 Completable 正在完成，但拒绝使用 flatMapCompletable？

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

注意userRoutingService.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 所做的：

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

使用 flatMapCompletable 时，返回的 Completable 会等待上游的 Observable 终端事件（onComplete）。

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

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

Answer 2

使用flatMapCompletable时，需要自己返回Completable.complete()。

编辑：

 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：因为disposableRule是Completable

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