RxJavaでレルムを使用する2

Question

私はAndroidアプリケーションでRxJava 2を使用しており、Realmを統合しています。私の知る限り、レルムはそうのように、デフォルトでRxJava 1をサポートし、RealmResults<?>を照会するときObservableを返すことができます。

Realm.getDefaultInstance() 
    .where(VideoBundle.class) 
    .findAll() 
    .asObservable() 
    .first() 

を観察可能で返された、私がレルムを使用するにはどうすればよいRxJava 1からですRxJava 2と一緒に？私は2つの関連する問題に遭遇しました。hereとhereが見つかりましたが、簡潔な答えは見つかりませんでした。さらに、ドキュメント（https://realm.io/docs/java/latest/#rxjava）にはカスタムRxObservableFactoryの作成が記載されていますが、その方法についての情報は提供されていません。

すでにRxJava 2を使用しているプロジェクトでRealmを使用するにはどうすればよいですか？

Answer 1

解決策は、最新のバックプレッシャー戦略を使用してRealmResultsをFlowableでラップすることです。上記レルム4.0.0-RC1から

private io.reactivex.Flowable<RealmResults<_>> getSomeItems() { 
    return io.reactivex.Flowable.create(new FlowableOnSubscribe<RealmResults<__>>() { 
     @Override 
     public void subscribe(FlowableEmitter<RealmResults<__>> emitter) 
       throws Exception { 
      Realm realm = Realm.getDefaultInstance(); 
      RealmResults<__> results = realm.where(__.class).findAllSortedAsync("__"); 

      final RealmChangeListener<RealmResults<__>> listener = _realm -> { 
       if(!emitter.isUnsubscribed() && results.isLoaded()) { 
        emitter.onNext(results); 
       } 
      }; 
      emitter.setDisposable(Disposables.fromRunnable(() -> { 
       results.removeChangeListener(listener); 
       realm.close(); 
      })); 
      results.addChangeListener(listener); 
     } 
    }, BackpressureStrategy.LATEST) 
    .subscribeOn(AndroidSchedulers.mainThread()) 
    .unsubscribeOn(AndroidSchedulers.mainThread()); 

---

、私は上記示したこの動作はrealmResults.asFlowable()を使用して焼成します。

Disposable subscription = realm.where(__.class) 
           .findAllSortedAsync("__") 
           .asFlowable() 
           .filter(RealmResults::isLoaded) 
           .subscribe(...);