Flowable<T> scroll(ObjectNode query) {
    final Map<String, String> params = Collections.singletonMap("scroll", keepAlive.getSeconds() + "s");
    final boolean hasLimit = query.has("size");
    final long limit = hasLimit ? query.get("size").asLong() : Long.MAX_VALUE;

    if (scrollSize > limit) {
      // don't use scrolling when batch size is greater than limit
      return ops.search(query, converter);
    }

    // override size during scrolling
    query.put("size", scrollSize);

    return ops.searchRaw(query, params)
            .toFlowable()
            .scan(AccumulatedResult.empty(limit), AccumulatedResult::next)
            .skip(1)// skip first empty accumulator
            .compose(transformer())
            .map(AccumulatedResult::result)
            .flatMapIterable(r -> r.searchHits().hits())
            .compose(f -> hasLimit ? f.limit(limit) : f) // limit number of elements
            .map(hit -> converter.convert(hit.source()));
  }
Duration to keep scroll context alive.
criteria/elasticsearch/src/org/immutables/criteria/elasticsearch/Scrolling.java