Batch loading with Retrofit and RxJava

Working on some project, I have find out, that there is no any standard solutions to implement batch-loading throw retrofit2 and retrofit2 adapter-rxjava bundle.
For example, have retrofit service interface.

interface Api {

     @GET("query.json")

     Observable<List<SomeEntityServerView>> getAll(

         @Query("first") int first,

         @Query("max") int batchSize);

}

To load single batch we do:

service.getAll(0,20).map(list -> ...).observeOn(...).subscribe(...);

To load next batch we have to create new observable every time. But we want to process all enities in single rx-java stream.

What about following simple approach?

public class BatchLoadingUtils {     

    /**

     * @param batchLoaderFactory - observable factory by batch number

     * @param batchSize

     */ 

    public static <T> Observable<List<T>> create(Func1<Integer, Observable<List<T>>> batchLoaderFactory, int batchSize) {

        //firstElementNumber

         AtomicInteger first = new AtomicInteger(0);

         BehaviorSubject<Integer> subject = BehaviorSubject.create(0);

         return subject

                 //map first id offset to observable

                 .flatMap(batchLoaderFactory::call)

                 .doOnNext(ts -> {

                     if (ts.size() == batchSize) {//Queue next batch in case current batch is complete

                          subject.onNext(first.addAndGet(batchSize));

                      } else {//Turning down, otherwise

                          subject.onCompleted();

                      }

                 });

     }

}

Usage example:

final int batchSize = 10; 

BatchLoader

     .create(

         first -> retrofitService.getAll(first, batchSize),

         batchSize

     )

     .observeOn(Schedulers.computation())

     .flatMapIterable(list -> list)

     .map(TimeEntryServerView::buildTimeEntry)

     .buffer(batchSize)

     .subscribe(...);

Looks like it is easy and effective. Now you can create single stream for several batches.