TaskWithRx.java 文件源码

@Override public void executeTask(final Ui ui, int totalTask) {
  List<Observable<ApiResponse>> calls = new ArrayList<>();
  for (int i = 0; i < totalTask; i++) {
    Observable<ApiResponse> apiResponseObservable = apiCall.callObservable(i + 1);
    Observable<ApiResponse> observableOnNewThread =
        apiResponseObservable.subscribeOn(Schedulers.newThread());
    calls.add(observableOnNewThread);
  }

  Observable.zip(calls, new FuncN<Long>() {
    @Override public Long call(Object... args) {
      return System.currentTimeMillis();
    }
  })
      .subscribeOn(Schedulers.newThread())
      .observeOn(AndroidSchedulers.mainThread())
      .subscribe(new Action1<Long>() {
        @Override public void call(Long time) {
          ui.showTime(time);
        }
      }, new Action1<Throwable>() {
        @Override public void call(Throwable throwable) {
          ui.showError("error " + throwable);
        }
      });
}