android - RxJava2- Completable永无止境

Question

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

基本上，我想进行api调用，然后将信息保存在数据库中，以便下次启动应用程序时从数据库加载数据。因此，为实现此目的，我为数据库中的每个插入创建了一个completable，然后每完成一次它就会在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") })
            }

然后在我的 Activity 中

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()