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.