Сделайте n асинхронных вызовов из списка и вызовите метод, когда все вызовы завершены

Question

Я делаю n асинхронных вызовов (n равен размеру массива и индексам, передаваемым в вызовах как целочисленные параметры) и хочу вызвать метод, когда все вызовы завершены.Я реализовал следующий код ниже.Я использовал счетчик, чтобы знать, что все звонки завершены.Это работает, однако я знаю, что это можно сделать более эффективным и элегантным способом.

int n = mUserUrls.getM3u().size();

counter = n;

Observable.range(0, n)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.doOnNext(new Consumer<Integer>() {
    @Override
    public void accept(Integer integer) throws Exception {
        final int index = integer;

        Single<ResponseBody> singleGetChannels = aPI.getChannels(mUserUrls.getM3u().get(integer))
                                    .subscribeOn(Schedulers.newThread())
                                    .observeOn(AndroidSchedulers.mainThread());

        Single<List<EPG>> singleGetEPGs = aPI.getEPGs(mUserUrls.getJson())
                                    .subscribeOn(Schedulers.newThread())
                                    .observeOn(AndroidSchedulers.mainThread());


        Single.zip(singleGetChannels, singleGetEPGs, new BiFunction<ResponseBody, List<EPG>, ChannelsAndEPG>() {
            @Override
            public ChannelsAndEPG apply(ResponseBody responseBodyChannels, List<EPG> ePGs) {

                ChannelsAndEPG channelsAndEPG = new ChannelsAndEPG(responseBodyChannels, ePGs);

                return channelsAndEPG;
            }
        }).subscribe(new SingleObserver<ChannelsAndEPG>() {
                                @Override
                                public void onSubscribe(Disposable d) {

                                }

                                @Override
                                public void onSuccess(ChannelsAndEPG channelsAndEPG) {
                                    m3Us.put(index, M3UParser.parseList(channelsAndEPG.mResponseBodyChannels.byteStream()));


                                    setEPGs(index, channelsAndEPG.mEPG);
                                    setEPGsForNext24Hours();

                                    counter--;
                                    if (counter == 0) {
                                        if (mCallback != null) {
                                            isDataLoaded = true;
                                            mCallback.onDataLoaded();
                                        }
                                    }

                                }

                                @Override
                                public void onError(Throwable e) {

                                }
                            });
                        }
                    })
                    .subscribe(new Observer<Integer>() {

                        @Override
                        public void onSubscribe(Disposable d) {

                        }

                        @Override
                        public void onNext(Integer integer) {

                        }

                        @Override
                        public void onError(Throwable e) {

                        }

                        @Override
                        public void onComplete() {
                            Log.i(TAG, "onComplete called");

                        }
                    });

Answer 1

Вы можете использовать это:

Observable.fromIterable(mUserUrls.getM3u())
          .flatMap{  m3u -> 
               aPI.getChannels(m3u.getInteger)
                  .zipWith(aPI.getEPGs(mUserUrls.getJson()))
                  .subscribeOn(Schedulers.io())
          }
          .doOnNext{
              m3Us.put(index, M3UParser.parseList(channelsAndEPG.mResponseBodyChannels.byteStream()));
              setEPGs(index, channelsAndEPG.mEPG);
              setEPGsForNext24Hours();
          }
          .subscribe(new Observer<Integer>() {

                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(Integer integer) {

                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onComplete() {
                        Log.i(TAG, "onComplete called");

                    }
                })

Answer 2

Вы можете использовать flatMap для преобразования каждого целого числа в одиночное (так же, как вы делаете это сейчас).А затем вызовите toList, чтобы получить Single.