将领域与 PublishSubject 一起使用答案

【问题标题】：Using realm with PublishSubject将领域与 PublishSubject 一起使用
【发布时间】：2018-09-05 21:26:32
【问题描述】：

我想将我的领域结果映射到一个不可变的视图模型，并且我想听结果的变化，所以我将它们发布为 PublishSubject，但是，在我旋转设备之前，数据不会出现在我的 recyclerview 中，当我删除 observeOn(AndroidSchedulers.mainThread()) 时，此问题已得到解决。

存储库：

fun notionsChanges(state: Boolean): Observable<Pair<MutableList<Notion>, OrderedCollectionChangeSet?>> {

        val notionsChanges = PublishSubject.create<Pair<MutableList<Notion>, OrderedCollectionChangeSet?>>()

        val realm = Realm.getDefaultInstance()
        val queryResult = realm.where<Notion>()
                .equalTo("isArchived", state)
                .findAllAsync()
        val listener: OrderedRealmCollectionChangeListener<RealmResults<Notion>> = OrderedRealmCollectionChangeListener { realmResults, changeSet ->
            if (realmResults.isValid && realmResults.isLoaded) {
                val results: MutableList<Notion> = realm.copyFromRealm(realmResults)
                notionsChanges.onNext(results to changeSet)
            }
        }
        queryResult.addChangeListener(listener)
        notionsChanges.doFinally {
            queryResult.removeChangeListener(listener)
            closeRealm(realm)
        }.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread())

        return notionsChanges
}

在我的演示者中，我使用这个 observable 将模型映射到视图模型，然后我在片段内的 recyclerview 中显示（订阅时）数据：

private var subscriptions: CompositeDisposable = CompositeDisposable()

override fun onResume() {
    super.onResume()
    showData()
}

override fun onPause() {
    subscriptions.clear()
    super.onPause()
}

private fun showData() {
        val viewModel = present(idleStates, resources, isIdle)
        with(viewModel) {
            subscriptions.addAll(
                    notionsChanges.subscribe(notionsAdapter::handleChanges),
                    //other subscriptions.
            )
        }
}

notionsAdapter.handleChanges：

fun handleChanges(collectionChange: Pair<List<NotionCompactViewModel>, OrderedCollectionChangeSet?>) {
    val (collection, changeset) = collectionChange
    debug("${collection.size}") //correctly prints the actual size of the collection.
    replaceAll(collection)
    if (changeset == null)
        notifyDataSetChanged()
    else {
        for (change in changeset.changeRanges)
            notifyItemRangeChanged(change.startIndex, change.length)

        for (insertion in changeset.insertionRanges)
            notifyItemRangeInserted(insertion.startIndex, insertion.length)

        for (deletion in changeset.deletionRanges)
            notifyItemRangeRemoved(deletion.startIndex, deletion.length)
    }
}

如果代码不清楚，请见谅。

edit：我的 onBindViewHolder 有时不会被调用（当然，当 recyclerview 为空时）。

【问题讨论】：

真正令人惊讶的是它完全有效。非循环线程不支持findAllAsync 和addChangeListener，这应该在从 HandlerThread 的循环线程创建的调度程序上运行。
即使我是从主线程调用它？
啊，你的subscribeOn(io())骗了我。那么这对于大型数据集来说是不安全的。哎呀！您确定不应该改用 BehaviorSubject 吗？
等一下。您使用的是 Realm 5.0+ 吗？变更集不再可以为空。你得到state == INITIAL。
是的，这就是我现在正在使用的，现在原始问题（使用 changeSet）已解决，您的其他具有干净架构的解决方案也正在运行，再次感谢。

标签： android realm rx-java rx-android publishsubject

【解决方案1】：

从 Realm 5.0 开始，初始变更集不再使用 changeset == null 发出信号。

您需要检查：

if(changeSet.getState() == State.INITIAL) {
    adapter.notifyDataSetChanged()

【讨论】：