如何從rxjava鏈中調用非可觀察代碼

Question

我在演示者中實現了一個rxjava調用鏈。 如果沒有從sqlite數據庫返回結果，它會調用遠程服務器並進行改進。

但是，一切都使用rxjava，除了我們的庫調用sqlite。當sqlite返回結果時，它會向我顯示3個結果爲空值。它似乎無法調用rxjava鏈中的非可觀察代碼？如何在不使用StorIO或Brite的情況下執行此操作？

結果看起來像這樣

演示

@Override 
protected void onCreate(Bundle savedState) { 
    super.onCreate(savedState); 

    restartableLatestCache(REQUEST_ASSIGNMENTS, 
     () -> mRepository.query() 
        .subscribeOn(Schedulers.io()) 
        .observeOn(mainThread()), 
     (assignmentActivity, response) -> assignmentActivity.onSuccess(response), 
     (assignmentActivity, throwable) -> assignmentActivity.onError(throwable) 
    ); 
} 

庫

return mAssignmentLocalDataStore.query(new AssignmentSpecification()) 
      .flatMap(assignments -> assignments == null || assignments.isEmpty() ? 
        mAssignmentRemoteDataStore.query() 
         .flatMap(remoteAssignments -> 
          Observable.zip(
            mEntityRepository.query() 
              .flatMap(mEntityRepository::add), 
            mFacilityRepository.query() 
              .flatMap(mFacilityRepository::add), 
            mAssignmentLocalDataStore.query(new AssignmentSpecification()), 
            (remoteEntities, remoteFacilities, assignmentsRetry) -> assignmentsRetry 
          ) 
         ): Observable.just(assignments) 
      ); 

SQLite的LocalDataStore

@Override 
public Observable<List<Assignment>> query(Specification specification) { 
    final SqlSpecification sqlSpecification = (SqlSpecification) specification; 

    final SQLiteDatabase database = mOpenHelper.getReadableDatabase(); 
    final List<Assignment> assignments = new ArrayList<>(); 

    try { 
     final Cursor cursor = database.rawQuery(sqlSpecification.toSqlQuery(), new String[]{}); 

     for (int i = 0, size = cursor.getCount(); i < size; i++) { 
      cursor.moveToPosition(i); 

      assignments.add(mToAssignmentMapper.map(cursor)); 
     } 

     cursor.close(); 

     return Observable.just(assignments); 

    } finally { 
     database.close(); 
    } 
} 

Answer 1

變化query爲類似以下內容（使用Observable.create）。另一個變化是返回Observable<Assignment>併爲每個記錄調用subscriber.onNext。

@Override 
public Observable<List<Assignment>> query(Specification specification) { 
    return Observable.create(subscriber -> { 
      final SqlSpecification sqlSpecification = (SqlSpecification) specification; 

      final SQLiteDatabase database = mOpenHelper.getReadableDatabase(); 
      final List<Assignment> assignments = new ArrayList<>(); 

      try { 
       final Cursor cursor = database.rawQuery(sqlSpecification.toSqlQuery(), new String[]{}); 

       for (int i = 0, size = cursor.getCount(); i < size; i++) { 
        cursor.moveToPosition(i); 

        assignments.add(mToAssignmentMapper.map(cursor)); 
       } 

       subscriber.onNext(assignments); 

       cursor.close(); 
      } finally { 
       database.close(); 
       subscriber.onCompleted(); 
      } 
    } 
}