RxJava. Борьба с вызовами суровой действительности

[7voprosov]

Те, кто уже с Rx «не на Вы» могут возразить мне, что для аккумулирования «выпущенных» элементов существует специальный оператор scan.

Не очень понятно почему вы пишите про scan, т.к по описанию вашей задаче кажется, что частичные суммы вам не нужны. Возможно, здесь уместнее говорить про reduce (который конечно является частным случаем для scan)

Зачем же городить огород из toList и map? Дело в том, что, если первый же запрос вернет нам пустой список (и тогда этот запрос будет единственным), scan выдаст ошибку

То же не совсем понятно, почему должна возникать ошибка, т.к здравая логика подсказывает (доки данный случай не обговаривают), что аккумулятор примет просто значение первого элемента (в данном случаи вашего пустого списка) и ни чего особенного не произойдет. Проверочный код в «лоб» работает:

List<String> empty = new ArrayList<String>();

Observable.just(empty).scan((accum, nextList) -> {
  accum.addAll(nextList);
  return accum;
}).subscribe(list -> {
  System.out.println("Should be empty list: " + list.size());
});

Я потратил немало сил и времени, чтобы написать такой простой с виду код.

В данном случае так же не понятно, почему не взять и просто заиспользовать ваш код из самого первого примера (характер ваших запросов говорит о их связанности, а значит все они все равно должны будут исполняться в одном потоке), как то так:

Future<List<StatementRUR>> future = executor.submit(
  (Callable<List<StatementRUR>>) () -> {
    int i = 0;
    List<StatementRUR> commonList = new ArrayList<>();
    while (true) {
      // Retrofit должен иметь простой сихронный доступ
        List<StatementRUR> list = 
          statementRURList(int i * limit, int limit, String accountId,
                           int periodDaysCount, Date docInfoDocDate);
        commonList.addAll(list);
        if (list.size() < limit) {
          break;
        }
        i++;
    }
    return commonList;
  }
);

[xoxol_89]

Добрый день! Спасибо за замечания.
Про reduce согласен. Он лучше подходит.
Про scan, я точно не помню уже, при каких конкретно условиях он падал (или выдавал исключение). По-моему, при условии, если в вышестоящем observable, только один «выпущенный» элемент. Нужно вспомнить будет)
Про «потратил силы» я скорее имел в виду, как это реализовать в Rx, потому что подобного не нашел.
Замечание про то, что можно использовать мой первый пример и отталкиваться от него, так как все необходимо будет выполнять только в одном фоновом потоке, отчасти согласен. В принципе можно. Но тут встает вопрос, что по циклу вы вот так запросы не сможете посылать. RetroFit ругнется и выкинет нас. Если использовать свой кастомный клиент, то все зависит от реализации очереди. И без пробрасывания колбэков не обойтись. А вот если использовать библиотеку Volley, которая возвращает RequestFuture, уже можно будет поиграться.
Но не стоит забывать про то, что код то в Rx выглядит проще. Уж Вы то со мной согласитесь) Плюс мы довольно легко сможем подключить, если необходимо, еще дополнительные операции и не переживать, что что-то отвалиться. Ну и обработка ошибок. В Rx она тоже удобнее, нежели в асинхронном коде.

[7voprosov]

Но тут встает вопрос, что по циклу вы вот так запросы не сможете посылать. RetroFit ругнется и выкинет нас.

Опять таки непонятно почему, на главной странице retrofit-а, написано:

Each Call from the created GitHubService can make a synchronous or asynchronous HTTP request to the remote webserver.

Так, что кажется, что проблем вообще ни каких не будет (исполнение идет в фиче) и колбэки так же не понадобится.

[xoxol_89]

Да, действительно, у RetroFit есть такое возвращаемое значение, как Call.
RetroFit ругнется, если вы вместо concatMap подставите flatMap, но это несколько другая история.
Но можно к примеру добавить параллельные вызовы в БД для записи или чтения, и уже добавляется настоящая асинхронность, и цикл синхронных вызовов уже не сработает так, как хотелось бы.

[lNevermore]

Хорошая статья, спасибо.

Согласен с 7voprosov про reduce, еще хотел отметить, что в примере у вас после concatMap идет map, и по сути map не нужен, можно вставить этот геттер прямо в concatMap, убрав лишний оверхэд и сделав код более коротким и более читабельным (на вкус и цвет, разумеется).

[xoxol_89]

Согласен по map. Спасибо)

[xoxol_89]

По поводу reduce, scan.
Вот какая ошибка может вылетать:

java.util.NoSuchElementException: Sequence contains no elements at rx.internal.operators.OperatorSingle$ParentSubscriber.onCompleted(OperatorSingle.java:131) at rx.internal.operators.OperatorTakeLastOne$ParentSubscriber.onCompleted(OperatorTakeLastOne.java:106) at rx.internal.operators.OperatorScan$2.onCompleted(OperatorScan.java:123) at rx.observers.SerializedObserver.onCompleted(SerializedObserver.java:99) at rx.observers.SerializedSubscriber.onCompleted(SerializedSubscriber.java:65) at rx.internal.operators.OperatorConcat$ConcatSubscriber.subscribeNext(OperatorConcat.java:171) at rx.internal.operators.OperatorConcat$ConcatSubscriber.onCompleted(OperatorConcat.java:154) at rx.internal.operators.OperatorMap$1.onCompleted(OperatorMap.java:44) at rx.observers.SerializedObserver.onCompleted(SerializedObserver.java:99) at rx.observers.SerializedSubscriber.onCompleted(SerializedSubscriber.java:65) at rx.internal.operators.OperatorConcat$ConcatSubscriber.subscribeNext(OperatorConcat.java:171) at rx.internal.operators.OperatorConcat$ConcatSubscriber.onCompleted(OperatorConcat.java:154) at rx.internal.operators.OperatorMap$1.onCompleted(OperatorMap.java:44) at rx.internal.operators.OperatorMap$1.onCompleted(OperatorMap.java:44) at rx.internal.operators.OperatorSingle$ParentSubscriber.onCompleted(OperatorSingle.java:125) at rx.internal.operators.OperatorTakeLastOne$ParentSubscriber.emit(OperatorTakeLastOne.java:158) at rx.internal.operators.OperatorTakeLastOne$ParentSubscriber.onCompleted(OperatorTakeLastOne.java:124) at rx.internal.operators.OperatorScan$2.onCompleted(OperatorScan.java:123) at rx.internal.operators.OperatorTakeUntilPredicate$ParentSubscriber.onNext(OperatorTakeUntilPredicate.java:57) at rx.internal.operators.OperatorMap$1.onNext(OperatorMap.java:55) at rx.observers.SerializedObserver.onNext(SerializedObserver.java:159) at rx.observers.SerializedSubscriber.onNext(SerializedSubscriber.java:95) at rx.internal.operators.OperatorConcat$ConcatInnerSubscriber.onNext(OperatorConcat.java:206) at retrofit.RxSupport$2.run(RxSupport.java:56) at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:442) at java.util.concurrent.FutureTask$Sync.innerRun(FutureTask.java:305) at java.util.concurrent.FutureTask.run(FutureTask.java:137) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1076) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:569) at java.lang.Thread.run(Thread.java:856)

Собственно поэтому используются комбинация операторов toList и map