RxJava：データベースとリモートサーバー

Question

ローカルデータベース（使用可能な場合）から取得するオブジェクトのリスト、またはリモートサーバーから取得するオブジェクトのリストがあります。私はRxJava Observables（データベースの場合はSqlBrite、リモートサーバーの場合はRetrofit）を使用しています。次のように

マイクエリコードは次のとおりです。

Observable<List<MyObject>> dbObservable = mDatabase 
      .createQuery(MyObject.TABLE_NAME,MyObject.SELECT_TYPE_A) 
      .mapToList(MyObject.LOCAL_MAPPER); 
Observable<List<MyObject>> remoteObservable = mRetrofitService.getMyObjectApiService().getMyObjects(); 

return Observable.concat(dbObservable, remoteObservable) 
    .first(new Func1<List<MyObject>, Boolean>() { 
       @Override 
       public Boolean call(List<MyObject> myObjects) { 
        return !myObjects.isEmpty(); 
       } 
      }); 

私はネットワークの要求がない、最初に観測可能な走りを見ると、空のリストを使用して最初の方法を打つが、その後改造、観察が実行されません。私が観測値の順番を入れ替えるか、リモート観測可能値を返すだけであれば、期待どおりに動作し、リモートサーバにヒットし、オブジェクトのリストを返します。

このシナリオでは、リモート観測可能オブジェクトが実行されないのはなぜですか？ Observableとdbを最初に連結して2番目のものを再構築すると、加入者のonNext、orErrorおよびonCompleteメソッドは呼び出されません。

ありがとうございます！

Answer 1

Kaushik Gopalは、彼のRxJava-Android-Samples githubプロジェクトでこれに対処しました。

彼はこの技術を使用することをお勧めします：

remoteObservable 
     .publish(network -> 
        Observable.merge(network, 
             dbObservable.takeUntil(network))) 
     .first(myObjects -> !myObjects.isEmpty()); 

編集：あなたはこの必要がある場合がありますようですね：

を

あなたの場合

getFreshNetworkData() 
      .publish(network -> 
         Observable.merge(network, 
              getCachedDiskData().takeUntil(network))) 
      .subscribeOn(Schedulers.io()) 
      .observeOn(AndroidSchedulers.mainThread()) 
      .subscribe(new Subscriber<List<MyObject>() { 

       ... 
      }); 

、それは次のようになります。

dbObservable 
    .flatMap(localResult -> { 
     if (localResult.isEmpty()) { 
      return remoteObservable; 
     } else { 
      return Observable.just(localResult); 
     } 
    }); 

Answer 2

私は、あなたが観測データを入手できると仮定しますあなたのローカルおよびリモート下記のようなから：

 final Observable<Page> localResult = mSearchLocalDataSource.search(query); 
     final Observable<Page> remoteResult = mSearchRemoteDataSource.search(query) 
       .doOnNext(new Action1<Page>() { 
        @Override 
        public void call(Page page) { 
         if (page != null) { 
          mSearchLocalDataSource.save(query, page); 
          mResultCache.put(query, page); 
         } 
        } 
       }); 

その後、あなたがそれらをマッピングし、リモート使用しない場合、ローカルであればローカル可能な利用を意味している最初の取得できます。

 return Observable.concat(localResult, remoteResult) 
       .first() 
       .map(new Func1<Page, Page>() { 
        @Override 
        public Page call(Page page) { 
         if (page == null) { 
          throw new NoSuchElementException("No result found!"); 
         } 
         return page; 
        } 
       }); 

を、以下のようにそれをサブスクライブ：

https://github.com/savepopulation/wikilight

：

mCompositeSubscription.clear(); 
     final Subscription subscription = mSearchRepository.search(this.mQuery) 
       .subscribeOn(Schedulers.io()) 
       .observeOn(AndroidSchedulers.mainThread()) 
       .subscribe(new Observer<Page>() { 
        @Override 
        public void onCompleted() { 
         // Completed 
        } 

        @Override 
        public void onError(Throwable e) { 
         mView.onDefaultMessage(e.getMessage()); 
        } 

        @Override 
        public void onNext(Page page) { 
         mView.onDefaultMessage(page.getContent()); 
        } 
       }); 

     mCompositeSubscription.add(subscription); 

詳細またはたとえば、あなたは私のgithubのレポをチェックすることができます

幸運を祈る！

編集：

次のようにローカルで観察できます。単にレコードがあるかどうかをチェックし、空のオブザーバブルを返します。

@Override 
public Observable<Page> search(@NonNull final String query) { 
    return Observable.create(new Observable.OnSubscribe<Page>() { 
     @Override 
     public void call(Subscriber<? super Page> subscriber) { 
      final Realm realm = Realm.getInstance(mRealmConfiguration); 
      final Page page = realm.where(Page.class) 
        .equalTo("query", query) 
        .findFirst(); 
      if (page != null && page.isLoaded() && page.isValid()) { 
       Log.i("data from", "realm"); 
       subscriber.onNext(realm.copyFromRealm(page)); 
      } else { 
       Observable.empty(); 
      } 
      subscriber.onCompleted(); 
      realm.close(); 
     } 
    }); 
} 

編集2：

あなたは地元の連結からnullを返すと、最初に動作しませんし、ヌルが観測リターンがnullの意味が、まだ観察することができますので、お使いのリモコンが呼び出されません。 observable.emptyをconcatで返すと、最初はobservableがローカルより多くのものを放出することができず、リモートからemitできることを意味します。