Android RxJava 订阅无处触发

Question

我有一个简单的应用程序，它显示从 API 中提取的联系人列表。
如果应用程序启动时没有网络，我需要显示最新的联系人获取。所以这就是我使用 Room 将联系人保存到数据库中的原因。

联系人已正确保存，需要时已正确拉出。
但是当我执行以下模式时出现了一个奇怪的问题：
- 我从 API 中提取了一些联系人（自动保存到本地数据库中）
- 我杀死了应用程序；
- 我切断所有网络以触发从本地数据库的拉取；
- 我在没有任何网络的情况下启动应用程序，联系人从本地数据库正确显示；
- 我打开网络以处理对 API 的新调用（清理数据库等...）
- API调用响应后，getContacts调用订阅后，调用getContactsFromDatabase的订阅！！

所以调试后我发现只是调用了订阅而不是完整的函数getContactsFromDatabase()，因为我在srList.isRefreshing = true 上的断点没有被触发。只有订阅部分的断点。

我还尝试在调用 getContactsFromDatabase 函数的唯一部分设置断点。它从未被触发，但是订阅中的断点被触发。

你可以在Github查看我的代码

ContactListFragment.kt：

    override fun onViewCreated(view: View, savedInstanceState: Bundle?) {
            super.onViewCreated(view, savedInstanceState)
            dataStorage = DataStorage(requireContext())
            initView()
            initListeners()
        }

        private fun initView() {
            val layoutManager = LinearLayoutManager(activity)
            rvContact.layoutManager = layoutManager
            rvContact.itemAnimator = DefaultItemAnimator()
            adapter = ContactListAdapter(this::onContactClicked)
            rvContact.adapter = adapter
            getContacts()
        }

        private fun initListeners() {
            srList.setOnRefreshListener {
                viewModel.page = 1; viewModel.contacts.clear(); getContacts()
            }

            rvContact.addOnScrollListener(object : RecyclerView.OnScrollListener() {
                override fun onScrolled(recyclerView: RecyclerView, dx: Int, dy: Int) {
                    super.onScrolled(recyclerView, dx, dy)
                    val manager = rvContact.layoutManager as LinearLayoutManager
                    val visibleItemCount: Int = manager.childCount
                    val totalItemCount: Int = manager.itemCount
                    val firstVisibleItemPosition: Int = manager.findFirstVisibleItemPosition()
                    if (!srList.isRefreshing) {
                        if ((visibleItemCount + firstVisibleItemPosition) >= totalItemCount
                            && firstVisibleItemPosition >= 0
                            && totalItemCount >= ContactListViewModel.PAGE_SIZE
                        ) {
                            getContacts()
                        }
                    }
                }
            })
        }

        private fun getContacts() {
            srList.isRefreshing = true
            disposable.add(viewModel.getContactByPages()
                .doOnError { e ->
                    srList.isRefreshing = false
                    if (viewModel.launch){
                        Timber.e("get contacts database")//breakpoint not triggered
                        getContactsFromDatabase()
                    }
                    e.localizedMessage?.let {
                        Timber.e(it)
                    }
                    val message = when (e) {
                        is BadRequestException -> {
                            getString(R.string.common_error_bad_request)
                        }
                        is ServerErrorException -> {
                            getString(R.string.common_error_server_error)
                        }
                        is UnknownHostException -> {
                            getString(R.string.common_error_no_connection)
                        }
                        else -> {
                            getString(R.string.common_error_basic)
                        }
                    }
                    Toast.makeText(requireContext(), message, Toast.LENGTH_SHORT).show()
                }
                .subscribe({ result ->
                    srList.isRefreshing = false
                    viewModel.page++
                    if (dataStorage.getBoolean(IS_FROM_CACHE)){//need a variable to clean the database after a first successful fetch
                        dataStorage.putBoolean(IS_FROM_CACHE, false).subscribe()
                        viewModel.contacts.clear()
                        cleanContactListOfDatabase()
                    }
                    viewModel.contacts.addAll(result.contacts)
                    saveContactsToDatabase()
                    adapter.updateList(viewModel.contacts)
                    tvNumberOfResult.text = getString(
                        R.string.contact_list_fragment_number_of_result,
                        viewModel.contacts.size
                    )
                }, Throwable::printStackTrace)
            )
        }

        private fun getContactsFromDatabase() {
            srList.isRefreshing = true//breakpoint not triggered here
            disposable.add(viewModel.getContactFromDatabase()
                .doOnError {
                    srList.isRefreshing = false
                }
                .subscribe({
                    srList.isRefreshing = false// breakpoint triggered here
                        viewModel.launch = false
                        viewModel.contacts.addAll(it)
                        adapter.updateList(viewModel.contacts)
                        tvNumberOfResult.text = getString(
                            R.string.contact_list_fragment_number_of_result,
                            viewModel.contacts.size
                        )
                        dataStorage.putBoolean(IS_FROM_CACHE, true).subscribe()
                }, Throwable::printStackTrace)
            )
        }

        private fun saveContactsToDatabase() {
            disposable.add(viewModel.insertAllContactsToDataBase()
                .doOnError {
                    Timber.e("Insert error")
                }
                .subscribe({
                    Timber.d("Contacts saved")
                }, Throwable::printStackTrace)
            )
        }

        private fun cleanContactListOfDatabase(){
            disposable.add(viewModel.cleanContactList()
                .doOnError {
                    Timber.e("clean table error")

                }
                .subscribe({
                    Timber.e("Table cleaned")
                }, Throwable::printStackTrace)
            )
        }

为了恢复问题，viewModel.getContactFromDatabase() 的订阅方法在没有调用函数getContactsFromDatabase() 的情况下被触发。
在没有任何网络的情况下打开应用程序（有本地数据库显示的联系人）；
打开任何网络（wifi 或 4g）；
尝试滑动刷新以从 API 中提取联系人；
getContacts()的订阅被触发（正常）；
viewModel.getContactFromDatabase() 的订阅被触发，甚至没有调用函数getContactsFromDatabase() -- 问题

你可以在Github查看我的代码

Answer 1

按照这些文档：

room/accessing-data#query-rxjava

room-rxjava-acb0cd4f3757

可流动/可观察

每次更新用户数据时，Flowable 对象都会自动发出，让您可以根据最新数据更新 UI。

在您的代码中，在getContactByPages() 之后调用saveContactsToDatabase() 将数据保存到数据库中

所以viewModel.getContactFromDatabase()（指向ContactDao.getContacts()）将再次发出数据。

如果您希望ContactDao.getContacts() 在被调用时只发出一次，请考虑将ContactDao.getContacts() 转换为MayBe/Single 而不是Observable

希望这会有所帮助。