Составьте несколько сетевых вызовов RxJava - Android

Question

Помогите в составлении множества сетевых вызовов и скопируйте результат в Rxjava. (Я использую в Android приложения.)

State 
-- List<City> cityList; 

City 
- cityId; 

RestCall 1 
Observable<State> stateRequest = restService.getStates(); 

RestCall 2 
Observable<CityDetail> cityRequest = restService.getCityDetail(cityId); 

В интерфейсе я должен отобразить список городов, после того, как получить все детали каждого города, а затем показать в ListView. Как достичь парсельных сетевых вызовов и накопить результат. ?

Я хочу, чтобы все детали детали города были помещены в Список в исходном состоянии 'объект'. Поскольку объект состояния имеет некоторую информацию, которая также должна быть удалена. Возможно ли это?

stateRequest ??? 
.subscribeOn(Schedulers.io()) 
.observeOn(AndroidSchedulers.mainThread()) 
.subscribe(new Subscriber<State>() { 
    @Override 
    public void onCompleted() { 

    } 

    @Override 
    public void onError(Throwable e) { 
    } 

    @Override 
    public void onNext(State result) { 
    // Get city list and display 
    } 
}); 

Я проверил этот пример, который показывает, как мы можем закрепить еще один наблюдаемый ответ. Ниже фрагмента показано 3 наблюдаемых вместе взятых. Но в моем случае я должен сделать 20 сетевых вызовов параллельными или последовательными (я имею в виду в фоновом режиме, но один за другим). Как мне это достичь. Любая помощь или указания?

https://gist.github.com/skehlet/9418379

Observable.zip(f3Observable, f4Observable, f5Observable, new Func3<String, Integer, Integer, Map<String, String>>() { 
    @Override 
    public Map<String, String> call(String s, Integer integer, Integer integer2) { 
     Map<String, String> map = new HashMap<String, String>(); 
     map.put("f3", s); 
     map.put("f4", String.valueOf(integer)); 
     map.put("f5", String.valueOf(integer2)); 
     return map; 
    } 

Answer 1

Я думаю, что ваш код может быть упрощен к чему-то вроде этого, как ваше использование оператора zip близок к использованию toList оператора

stateRequest 
.subscribe(State state -> { 
    Observable.from(state.getCityList()) 
       .flatMap(City city -> restService.getCityDetail(city.getId()) 
       .toList() 
       .subscribe(List<City> cities -> { 

        state.clear(); 
        state.addAll(cities); 
       }); 
    }); 

Как RxJava не обеспечивает throttle оператора, вы может построить нечто подобное:

Observable<City> limiter = Observable.zip(Observable.interval(1, SECONDS), aCity, (i, c) -> c); 

Использование этого ограничителя является наблюдаемым, который будет излучать город каждую секунду.

Так, с кодом, если вы хотите ограничить вызов getCityDetail, например:

Observable<Object> limiter = Observable.interval(1, SECONDS); 
stateRequest 
.subscribe(State state -> { 
    Observable.zip(limiter, Observable.from(state.getCityList()), (i, c) -> c) 
       .flatMap(City city -> restService.getCityDetail(city.getId()) 
       .toList() 
       .subscribe(List<City> cities -> { 

        state.clear(); 
        state.addAll(cities); 
       }); 
    }); 

Answer 2

stateRequest 
.flatMap(new Func1<State, Observable<State>>() { 
    @Override 
    public Observable<State> call(final State state) { 

     List<Observable> cityObservablesList = new ArrayList<Observable>(); 

     for(City city: state.getCityList()) { 
      cityObservablesList.add(restService.getCityDetail(city.getId()); 
     } 

     Observable cityObservables = Observable.from(cityObservablesList); 
     return Observables.zip(cityObservables, new FuncN<State>() { 
      @Override 
      public State call(Object... args) { 
       List<City> cityList = state.getCityList(); 
       cityList.clear(); 
       for(Object object: args) { 
        cityList.add((City)object); 
       } 

       return state; 
      } 
     }) 
    }) 
.subscribeOn(Schedulers.io()) 
.observeOn(AndroidSchedulers.mainThread()) 
.subscribe(new Subscriber<State>() { 
    @Override 
    public void onCompleted() { 

    } 

    @Override 
    public void onError(Throwable e) { 
    } 

    @Override 
    public void onNext(State result) { 
    // Get city list and display 
    } 
}); 

Я получил это работает с помощью почтового оператора и Iterable для списка городов в качестве первого параметра. Но я сталкиваюсь с другой проблемой. Поскольку zip выполняет задание параллельно, 10-15 сетевых вызовов выполняются параллельно и отбрасываются сервером с максимальной ошибкой Query Per Second (QPS-403). Как мне поручить zip-оператору выполнять задачи один за другим?

Я решил решить эту проблему, добавив задержку [c * 200, TimeUnit.MILLISECONDS))] к наблюдаемому городу. Но это не похоже на правильное решение.

Советуете?

Answer 3

Посмотрите на flatMap (функция .., BiFunction ..). Возможно, это то, что вам нужно.

statesRepository.getStates() 
    .flatMap(states-> Observable.fromIterable(states)) 
    .flatMap(
      state-> cityRepository.getStateCities(state), 
      (state, cityList) -> { 
       state.setCities(cityList); 
       return state; 
      }) 
    .subscribe(state-> showStateWithCity(state));