rxjavaチェーンから観測不可能なコードを呼び出す方法

Question

私はプレゼンターでrxjava呼び出しチェーンを実装しました。 sqliteデータベースから返された結果がない場合は、改造したリモートサーバーを呼び出します。

しかし、すべてsqliteを呼び出すリポジトリを除いてrxjavaを使用しています。 sqliteが結果を返すと、null値を持つ例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) 
      ); 

のように見えます（ Observable.createを使用して）次のように

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