如何在 RxJava2 中使用领域

Question

我在我的项目中使用 RxJava 2 和 Realm，并且我正在使用 Observable 从 RealmResults 中获取数据。这是我的代码。

fun getAllAsync(){
    realm?.where(RealmSuggestedFriends::class.java)
            ?.findAllAsync()
            ?.asFlowable()
            ?.filter { t -> t.isLoaded }
            ?.observeOn(AndroidSchedulers.mainThread())
            ?.subscribe({
        t: RealmResults<RealmSuggestedFriends>? ->for (firstResults in t!!){
        val requestPojo = RequestPojo()
        requestPojo.email = firstResults.friendEmail
        requestPojo.image = firstResults.friendImage
        requestPojo.name = firstResults.friendName
        requestPojo.status = firstResults.friendStatus
        requestPojo.thumb_image = firstResults.friendThumbImage
        requestPojo.uid = firstResults.friendUid
        userAdapter?.addData(requestPojo)
    }
    },{
        t: Throwable? ->
      },
      {},
      {})
    }

我尝试了上面的代码，但没有得到任何数据。在此之后，我尝试通过这种方式获取数据

Observable.create<RealmSuggestedFriends> { emitter ->
        val results = realm?.where(RealmSuggestedFriends::class.java)
                ?.findAllAsync()
        for (suggestedFriends in results!!){
            emitter.onNext(suggestedFriends)
        }
        emitter.onComplete()
    }.subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .flatMap { firstResults ->
                val requestPojo = RequestPojo()
                requestPojo.email = firstResults.friendEmail
                requestPojo.image = firstResults.friendImage
                requestPojo.name = firstResults.friendName
                requestPojo.status = firstResults.friendStatus
                requestPojo.thumb_image = firstResults.friendThumbImage
                requestPojo.uid = firstResults.friendUid
                Observable.just(requestPojo).subscribeOn(Schedulers.io())
            }
            .subscribe({
                t: RequestPojo? ->  userAdapter?.addData(t!!)
            },{
                t: Throwable? ->  
            },{

            },{
                t: Disposable? ->  compositeDisposable?.add(d!!)
            })

也使用此代码，我没有得到任何数据，而且我得到了 Realm accessed from incorrect thread 的祝酒词。任何人都可以帮助我使其工作。提前致谢。

Answer 1

Single.create<List<RealmSuggestedFriends>> { emitter ->
    Realm.getDefaultInstance().use { realm ->
        val results = realm.where(RealmSuggestedFriends::class.java).findAll()
        emitter.onSuccess(realm.copyFromRealm(results))
    }
}.subscribeOn(Schedulers.io())
 .observeOn(AndroidSchedulers.mainThread())
 .map { results ->
     for(firstResults in results) {
        val requestPojo = RequestPojo()
        requestPojo.email = firstResults.friendEmail
        requestPojo.image = firstResults.friendImage
        requestPojo.name = firstResults.friendName
        requestPojo.status = firstResults.friendStatus
        requestPojo.thumb_image = firstResults.friendThumbImage
        requestPojo.uid = firstResults.friendUid
     }
 }.subscribe({    
     list: List<RequestPojo> ->  userAdapter.updateData(list)
 },{
     t: Throwable? ->  
 },{
 })

无论如何，假设您不想在领域更改时收到通知。

否则，使用 Observable，有点像 this one：

private Observable<List<Task>> createResults(QuerySelector<DbTask> querySelector) {
    return Observable.create((ObservableOnSubscribe<List<Task>>) emitter -> {
        Realm realm = Realm.getDefaultInstance();
        final RealmResults<DbTask> dbTasks = querySelector.createQuery(realm);
        final RealmChangeListener<RealmResults<DbTask>> realmChangeListener = element -> {
            if(element.isLoaded() && !emitter.isDisposed()) {
                List<Task> tasks = mapFrom(element);
                if(!emitter.isDisposed()) {
                    emitter.onNext(tasks);
                }
            }
        };
        emitter.setDisposable(Disposables.fromAction(() -> {
            if(dbTasks.isValid()) {
                dbTasks.removeChangeListener(realmChangeListener);
            }
            realm.close();
        }));
        dbTasks.addChangeListener(realmChangeListener);
    }).subscribeOn(looperScheduler.getScheduler()).unsubscribeOn(looperScheduler.getScheduler());
}

要让第二个工作，您甚至可以尝试my library that helps with this。

Flowable<List<Dog>> dogs;

@Inject
Monarchy monarchy;

private List<Dog> currentDogs = Collections.emptyList();

@Override
public void onCreate(@Nullable Bundle savedInstanceState) {
    super.onCreate(savedInstanceState);
    LiveData<List<Dog>> dogs = monarchy.findAllMappedWithChanges(realm -> realm.where(RealmDog.class),
                                                           from -> Dog.create(from.getName()));
    this.dogs = Flowable.fromPublisher(LiveDataReactiveStreams.toPublisher(getActivity(), dogs));