如何結合兩個幾乎相似的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)); 
} 

，你可以直接merge之後添加.distinctUntilChange()避免雙重處理api.requestAPI1() res ULT。

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