用RxJava实现事务中复制对象的正确方法

Question

当前，当将值复制到领域时，我会执行以下操作：

public void addToRealm(Home item, RealmChangeListener<E> listener) {
    realm.executeTransaction((Realm realm1) ->
            realm1.copyToRealm(item).addChangeListener<Home>(listener));
}

然后我可以访问侦听器中新添加的对象。RxJava实现相同功能的正确方式是什么？可观察到的必须返回Observable<Home>，它是realmCopy而不是原始对象。任何人都可以提供样品吗？

通过这样做设法使其发挥作用，尽管我不确定这是最好的方法。。。建议的方法是什么？

return Observable.just(homeItem)
                        .map { (HomeItem homeItem) ->
                            return AnotherHomeItem(homeItem.xxx, homeItem.yyy)
                        }
                        .flatMap { (AnotherHomeItem anotherItem) -> 
                            realm.beginTransaction()
                            val newItem = realm.copyToRealm(anotherItem).asObservable< AnotherHomeItem >()
                            realm.commitTransaction()
                            return newItem
                        }
                        .filter {
                            return it.isLoaded
                        }

Answer 1

您应该在后台线程上写入Realm，并在UI线程上使用不同的订阅进行观察。

你在后台线程上坚持一个订阅：

public Subscription downloadObjectsFromNetwork() {
    return objectApi.getObjects()
        .subscribeOn(Schedulers.io())
        .subscribe(response -> {
            try(Realm realmInstance = Realm.getDefaultInstance()) {
                realmInstance.executeTransaction(realm -> realm.insertOrUpdate(response.objects));
            }
        });
}

你在UI线程上用asObservable()阅读：

public Subscription readFromRealm() {
    return realm.where(SomeObject.class)
         .findAllAsync()
         .asObservable()
         .subscribeOn(AndroidSchedulers.mainThread())
         .observeOn(AndroidSchedulers.mainThread())
         .filter(RealmResults::isLoaded)
         .subscribe(objects -> adapter.updateData(objects));
}

---

使用RxJava的Realm

对于查询，Realm提供了realmResults.asObservable()方法。观察结果只能在活套线程（通常是UI线程）上进行。

要使其工作，您的配置必须包含以下

realmConfiguration = new RealmConfiguration.Builder(context)       //
                          .rxFactory(new RealmObservableFactory()) //
                             //...
                          .build();

之后，您可以将您的结果用作可观察结果。

Observable<RealmResults<SomeObject>> observable = results.asObservable();

对于异步查询，应按isLoaded()筛选结果，以便仅在执行查询时才收到事件。同步查询不需要此filter()（isLoaded()在同步查询时总是返回true）。

    Subscription subscription = RxTextView.textChanges(editText).switchMap(charSequence -> 
        realm.where(SomeObject.class)
             .contains("searchField", charSequence.toString(), Case.INSENSITIVE)
             .findAllAsync()
             .asObservable())
    .filter(RealmResults::isLoaded) //
    .subscribe(objects -> adapter.updateData(objects));

对于写入，您应该使用executeTransactionAsync()方法，或者在后台线程上打开Realm实例，同步执行事务，然后关闭Realm实例。

public Subscription loadObjectsFromNetwork() {
    return objectApi.getObjects()
        .subscribeOn(Schedulers.io())
        .subscribe(response -> {
            try(Realm realmInstance = Realm.getDefaultInstance()) {
                realmInstance.executeTransaction(realm -> realm.insertOrUpdate(response.objects));
            }
        });
}