RxJava2とRetrofit2でパラレルに複数のノンブロッキングサービスリクエストを作成するには

Question

RxJava2を使用して並列非同期コールを実装する際には、何か助けが必要です。Retrofit2 & 私の要件は次のとおりです。

1）私は複数の保険会社を抱えています（現在は2つしかありません）。その保険会社名を使用して複数のパラレルリクエストを送信する必要があります。

2）いずれかのサーバーエラーが発生した場合、残りのリクエストはブロックされません。

以下は私が今まで試したことです。

ArrayList<String> arrInsurer = new ArrayList<>(); 
     arrInsurer.add(AppConstant.HDFC); 
     arrInsurer.add(AppConstant.ITGI); 

     RequestInterface service = getService(ServiceAPI.CAR_BASE_URL); 
     for (String insurerName : arrInsurer) { 
      service.viewQuote(Utils.getPrefQuoteId(QuoteListActivity.this), insurerName) 
        .subscribeOn(Schedulers.computation()) 
        .observeOn(AndroidSchedulers.mainThread()) 
        .subscribe(new Consumer<ViewQuoteResDTO>() { 
         @Override 
         public void accept(@NonNull ViewQuoteResDTO viewQuoteResDTO) throws Exception { 
          Log.e("Demo", viewQuoteResDTO.getPremiumData().getIDV()+""); 
          updateList(); 
         } 
        }, new Consumer<Throwable>() { 
         @Override 
         public void accept(@NonNull Throwable throwable) throws Exception { 
          Log.e("Demo", throwable.getMessage()); 
         } 
        }); 
     } 

private RequestInterface getService(String baseUrl) {  
    Gson gson = new GsonBuilder() 
      .setLenient() 
      .create(); 

    return new Retrofit.Builder() 
      .baseUrl(baseUrl)     
      .addCallAdapterFactory(RxJava2CallAdapterFactory.create()) 
      .addConverterFactory(GsonConverterFactory.create(gson)) 
      .build().create(RequestInterface.class); 

} 

上記のコードは、両方のリクエストが正常に応答した場合にのみ正常に動作します。しかし、いずれかの要求が内部サーバーエラーとして応答を与えると、残りの要求もブロックされます。

次のいずれかの要求が失敗した場合にログエラーが発生します。

E/Demo: HTTP 500 Aww Snap, Some thing happened at server. Please try back again later. 
E/Demo: unexpected end of stream on Connection{100.xxx.xxx.xx:portNo, proxy=DIRECT@ hostAddress=/100.xxx.xxx.xx:portNo cipherSuite=none protocol=http/1.1} 

このエラーを処理するにはどうすればよいですか？

Answer 1

他のRx関連の質問と同様に、これは複数の回答があります。私はあなたの私のアプリケーションで使用し、正確にこのユースケースを解決する私に与えます。それが役に立てば幸い。

ショートバージョン - これはmergeDelayErrorに依存しています。それをチェックしてくださいhere

なぜmergeですか？ concatとは異なり、オブザーバブルを並行して実行します。なぜmergeDelayErrorですか？それはエラーを遅らせる...本質的にすべての観測を実行し、すべてが終了したときにエラーを出す。これにより、1つまたは複数のエラーが発生しても、他のエラーは引き続き実行されます。

細部には注意が必要です。イベントの順序はもはや保持されません。つまり、mergeオペレータが観測可能なイベントの一部をインターリーブする可能性があります（あなたが以前のやり方を考えれば、これは問題ではありません）。私が知る限り、複数のオブザーバブルが失敗しても、onErrorコールが1つだけ得られます。これらの両方がOKであれば、あなたは次のことを試みることができる：

List<Observable<ViewQuoteResDTO>> observables = new ArrayList<>(); 
for (String insurerName : arrInsurer) {  
    observables.add(service.viewQuote(
     Utils.getPrefQuoteId(QuoteListActivity.this), insurerName)); 
} 

Observable.mergeDelayError(observables) 
      .subscribeOn(Schedulers.computation()) 
      .observeOn(AndroidSchedulers.mainThread()) 
      .subscribe(/* subscriber calls if you need them */); 

アイデアは、あなたが実行し、それらをトリガーするmergeDelayErrorを使用するつもりのすべての観測を作成することです。