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(); 
       } 
      }); 

我看到的第一個觀察的跑步和打一個空列表中的第一種方法，但後來改裝觀察到的不運行，沒有網絡的要求。如果我切換可觀察對象的順序，或者只是返回遠程可觀察對象，它會按預期工作，它會碰到遠程服務器並返回對象列表。

爲什麼在這種情況下遠程可觀察性無法運行？訂閱者的onNext，orError和onComplete方法不會在我將第一個數據庫和第二個數據庫連接起來時調用。

謝謝！

Answer 1

Kaushik Gopal已經在他的RxJava-Android-Samples github項目中解決了這個問題。

他建議使用這一技術：

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

       ... 
      }); 

在你的情況下，它可能是這樣的：

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

編輯：這聽起來像你也許會需要這樣：

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

Answer 2

我假設你有可觀察到的可以得到dat一個來自本地和遠程象下面這樣：

 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; 
        } 
       }); 

，並訂閱它象下面這樣：

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回購： https://github.com/savepopulation/wikilight

祝你好運！

編輯：

你可以試試當地的觀察到如下圖所示。只需檢查是否有記錄並返回空的可觀察值。

@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：

當您從本地CONCAT返回null，第一是行不通的，你的遙控器將不會被調用，因爲空是指觀察到的回報空，但仍然可以觀察到。當你用concat返回observable.empty時，首先這意味着observable不能從本地發出任何東西，因此它可以從遠程發出。