RxJava2: Как улучшить параллельную загрузку данных и кеширование?

Question

Я борюсь своим путем через RxJava2. Мне интересно, приемлемо ли мое решение или есть ли способы его улучшить.

USECASE

1. Кнопка Пользователь нажимает обновление данных
2. Откроется диалоговое окно - пожалуйста, подождите
3. Несколько звонков Серверные обрабатываются параллельно
4. Как только любой из этих вариантов отделки - данные сохраняются в локальной базе данных
5. После завершения всех запросов (обратный вызов и сохранение) диалог должен быть отклонен

Текущее решение

У меня есть несколько Completables глядя, как это:

Completable organisationUnitCompletable = backendService.getOrganisationUnits() 
    .doOnNext(data -> organisationUnitDao.saveInTx(data)) 
    .ignoreElements() 
    .subscribeOn(Schedulers.io()); 

Completable locationCompletable = backendService.getLocations() 
    .doOnNext(data -> locationDao.saveInTx(data)) 
    .ignoreElements() 
    .subscribeOn(Schedulers.io()); 

Completable prioritiesCompletable = backendService.getPriorities() 
    .doOnNext(data -> priorityDao.saveInTx(data)) 
    .ignoreElements() 
    .subscribeOn(Schedulers.io()); 

Я упаковать их в один путем добавления в список и с помощью merge оператора:

List<Completable> compatibles = new ArrayList<>(); 
compatibles.add(organisationUnitCompletable); 
compatibles.add(locationCompletable); 
compatibles.add(prioritiesCompletable); 

Completable.merge(compatibles) 
.subscribeOn(Schedulers.io()) 
.observeOn(AndroidSchedulers.mainThread()) 
.subscribe(() -> { 
    progressDialog.dismiss(); 
}); 

Возможная импровизация nts

Хорошо, так что это работает по назначению. Но есть вещи, о которых я не очень доволен.

Должен ли я добавить subscribeOn(Schedulers.io()) к каждому заполненному? Он не работает параллельно без него, но, может быть, есть лучший способ сделать это?

Все готовые изделия имеют следующие линии.

.ignoreElements() 
    .subscribeOn(Schedulers.io()); 

Есть ли способ извлечь его в один метод? Я пробовал что-то вроде этого:

private <T> Completable prepareCompletable(Function<Void, Observable<List<T>>> source, AbstractDao<T, Long> dao) { 

    Completable orderTypeCompletable = source 
      .doOnNext(data -> dao.saveInTx(data)) 
      .ignoreElements() 
      .subscribeOn(Schedulers.io()); 
} 

, где я бы просто положил Наблюдаемый и DAO. Конечно, он не компилируется. Похоже, для этого требуется гораздо больше знаний о дженериках, чем у меня уже есть.

Извините за длинный вопрос, трудно объяснить весь текст в нескольких словах.

Answer 1

Нужно ли мне добавлять subscribeOn (Schedulers.io()) к каждому заполненному?

Да, но он не нужен после Completable.merge().

Есть ли способ извлечь его в один метод?

public static <T> Function<Flowable<T>, Completable> applyIgnore() { 
    return f -> f.ignoreElements().subscribeOn(Schedulers.io()); 
} 

Completable locationCompletable = backendService.getLocations() 
.doOnNext(data -> locationDao.saveInTx(data)) 
.to(applyIgnore());