Как совместить два почти похожих метода RxJava?

Question

У меня есть два метода RxJava стиле:

void requestFirst() { 
    Observable 
     .combineLatest(
      api.requestAPI1(), 
      api.requestAPI2(), 
      (result1, result2) -> result1) 
     .map(result -> processResult(result)) 
     .observeOn(AndroidSchedulers.mainThread()) 
     .doOnSubscribe(etc1()) 
     .doOnSubscribe(etc2()) 
     .subscribe(
      result -> onComplete(result), 
      e -> onError(e)); 
} 

void requestNext() { 
    api.requestAPI1() 
     .map(result -> processResult(result)) 
     .observeOn(AndroidSchedulers.mainThread()) 
     .doOnSubscribe(etc1()) 
     .doOnSubscribe(etc2()) 
     .subscribe(
      result -> onComplete(result), 
      e -> onError(e)); 
} 

два метода являются почти то же самое, за исключением combineLatest, и я хочу, чтобы объединить его для рефакторинга. Итак, я пишу следующий код:

void request(boolean isFirst) { 
    Observable 
     .combineLatest(
      api.requestAPI1(), 
      (isFirst) ? api.requestAPI2() : Observable.empty(), 
      (result1, result2) -> result1) 
     .map(result -> processResult(result)) 
     .observeOn(AndroidSchedulers.mainThread()) 
     .doOnSubscribe(etc1()) 
     .doOnSubscribe(etc2()) 
     .subscribe(
      result -> onComplete(result), 
      e -> onError(e)); 
} 

Но я думаю, что второй параметр метода combineLatest не стиль Rx в. Как я могу объединить эти два похожих метода по стилю Rx? Спасибо.

Answer 1

Использование оператора .merge()

final Observable tworequestsObservable = Observable 
    .combineLatest(
     api.requestAPI1(), 
     api.requestAPI2(), 
     (result1, result2) -> result1) 

void requestNext() { 
    Observable.merge(api.requestAPI1(), tworequestsObservable) 
     .map(result -> processResult(result)) 
     .observeOn(AndroidSchedulers.mainThread()) 
     .doOnSubscribe(etc1()) 
     .doOnSubscribe(etc2()) 
     .subscribe(
      result -> onComplete(result), 
      e -> onError(e)); 
} 

И вы можете добавить .distinctUntilChange() только после того, как merge избежать двойная обработка api.requestAPI1() res ии.

Answer 2

Вы можете добавить новый метод:

void processResults(Observable<Source> dataSource) { 
    results.map(result -> processResult(result)) 
    .observeOn(AndroidSchedulers.mainThread()) 
    .doOnSubscribe(etc1()) 
    .doOnSubscribe(etc2()) 
    .subscribe(
     result -> onComplete(result), 
     e -> onError(e)); 
} 

и использовать его как:

void requestNext() { 
    processResults(api.requestAPI1()); 
}