Несколько вызовов api для Android с использованием RxJava

Question

У меня есть список объектов, которые отправляются как тело для запроса по почте в каждом вызове api. Все запросы на отправку должны выполняться параллельно. Для каждого запроса он должен уведомить об успешном завершении или неудаче, и после того, как все последующие сообщения будут завершены, он должен уведомить об этом. Как достичь этого с помощью rxjava

Пожалуйста, ознакомьтесь с этим кодом. Это реализация?

FlashSaleRetrofitService retrofitService = flashSaleProvider.createRetrofitService(); 
    FlashSale runningFlashSale = JumkeyApplication.getRunningFlashSale(); 

    List<Observable<Response<ClaimWrapper>>> wrapperResponseList = new ArrayList<>(); 

    for (final CartProduct cartProduct : cartProducts) { 

     final Claim claim = new Claim(); 

     Customer customer = customerDataProvider.getCustomer(); 
     claim.setCustomerId(String.valueOf(customer.getId())); 
     claim.setEmail(customer.getEmail()); 
     claim.setMobileNumber(customer.getDefaultAddress().getPhone()); 
     claim.setProductId(Long.parseLong(cartProduct.getProductId())); 
     claim.setVariantId(Long.parseLong(cartProduct.getVariantId())); 
     claim.setQuantity(cartProduct.getQuantity()); 
     claim.setSaleId(runningFlashSale.getSaleId()); 

     wrapperResponseList.add(retrofitService.claimProduct(claim)); 
    } 

    Observable.zip(wrapperResponseList, new FuncN<List<Claim>>() { 
     @Override 
     public List<Claim> call(Object... args) { 
      return null; 
     } 
    }).subscribeOn(Schedulers.newThread()) 
      .observeOn(AndroidSchedulers.mainThread()) 
      .doOnNext(new Action1<List<Claim>>() { 
       @Override 
       public void call(List<Claim> claims) { 

       } 
      }) 
      .compose(new Observable.Transformer<List<Claim>, Claim>() { 
       @Override 
       public Observable<Claim> call(Observable<List<Claim>> listObservable) { 
        return null; 
       } 
      }) 
      .subscribe(new Observer<Claim>() { 
       @Override 
       public void onCompleted() { 

       } 

       @Override 
       public void onError(Throwable e) { 

       } 

       @Override 
       public void onNext(Claim claim) { 

       } 
      }); 

Если это неправильный способ, предложите мне решить эту проблему.

Answer 1

Вы ищете метод Zip.

См. RxJava documentation.

Answer 2

Если вы хотите получить результаты процесса, когда все сделано.

Observable 
     .zip(observables, new FuncN<List<Response<ClaimWrapper>>>() { 
      @Override 
      public List<String> call(Object... args) { 
       List<Response<ClaimWrapper>> result = new ArrayList<>(args.length); 

       for (Object arg : args) { 
        result.add((Response<ClaimWrapper>) arg); 
       } 
       return result; 
      } 
     }) 
     .subscribe(new Subscriber<List<Response<ClaimWrapper>>>() { 
      @Override 
      public void onCompleted() { 

      } 

      @Override 
      public void onError(Throwable e) { 

      } 

      @Override 
      public void onNext(List<Response<ClaimWrapper>> results) { 

      } 
     }); 

Если вы хотите обработать один за другим, когда каждый его

Observable 
     .merge(observables) 
     .subscribe(new Subscriber<Response<ClaimWrapper>>() { 
      @Override 
      public void onCompleted() { 

      } 

      @Override 
      public void onError(Throwable e) { 

      } 

      @Override 
      public void onNext(Response<ClaimWrapper> result) { 

      } 
     });