RxJava2- 可完成永无止境

Question

我是 rxjava 世界的新手，正在尝试实现以下场景。

基本上，我想进行 api 调用，然后将信息保存在数据库中，以便下次启动应用程序时，它会从数据库中加载数据。因此，为了实现这一点，我为数据库中的每个插入创建了一个可完成的对象，然后每当它完成时，它都会在 UI 中通知我。但问题是它永远不会结束。

我的 api 调用：

fun getConversation(): Single<ConversationResponse>

它返回一个用户列表和一个消息列表

我在数据库中的插入：

fun insertUsers(users: List<User>) = Completable
        .create { usersDao.addUsers(users) }

fun insertMessages(messages: List<Message>) = Completable
        .create { messagesDao.addMessages(messages) }

然后我调用 api 并插入数据

fun getMessagesFromApi(): Completable {

    return conversationService.getConversation()
            .flatMapCompletable {
                conversationCache.insertUsers(it.users)
                        .doOnComplete { Log.d("ConversationActivity", "Insert users") }
                        .andThen(conversationCache.insertMessages(it.messages)
                                .doOnComplete { Log.d("ConversationActivity", "Insert messages") })
            }

然后在我的活动中

conversationRepository.getMessagesFromApi()
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(object : DisposableCompletableObserver() {
                override fun onStart() {
                    Log.d("ConversationActivity", "onStart")
                }

                override fun onComplete() {
                    Log.d("ConversationActivity", "onComplete")
                    mvpView?.loadedInfo()
                }

                override fun onError(e: Throwable) {
                    Log.d("ConversationActivity", "onError: $e.message")
                }
            })

在我的日志中我只看到D/ConversationActivity: onStart

为什么会这样？有人可以给我一些提示或告诉我我做错了什么吗？会很有帮助！！！

Answer 1

如果您使用Completable.create()，您需要在传递给函数的块内手动触发onComplete()|onError()。 在您的情况下，您可以使用 Completable.fromCallable() 或 Completable.fromAction()