observable pagination

paginatedobservable.java

@Test
    public void testPaginatedCalls() throws Exception {

        Observable<Response> call = createPaginatedCall(-1L, 3L);

        Observable<List<Long>> pages = flattenPages(call);
        // Verify:
        List<List<Long>> paginated = pages.toList().toBlocking().first();
        assertEquals(paginated, Arrays.asList(Arrays.asList(1L, 2L, 3L), Arrays.asList(4L, 5L)));

    }

    private static Observable<List<Long>> flattenPages(Observable<Response> responseObservable) {

        if (responseObservable == null) {
            return Observable.empty();
        } else {
            return responseObservable.flatMap(new Func1<Response, Observable<List<Long>>>() {
                @Override
                public Observable<List<Long>> call(Response response) {

                    Observable<Response> next = response.next();
                    List<Long> data = response.data();
                    return Observable.just(data).concatWith(flattenPages(next));
                }
            });
        }
    }

    private static Observable<Response> createPaginatedCall(Long maxId, Long perPage) {

        Response theResponse = null;

        if (maxId == -1) {
            //first page
            //load the data
            List<Long> firstPage = Arrays.asList(1L, 2L, 3L);
            Long updatedMaxId = firstPage.get(firstPage.size() - 1);
            //recur
            theResponse = new DemoResponse(firstPage, createPaginatedCall(updatedMaxId, perPage));
        } else {
            //next pages

            //load the data
            List<Long> lastPage = Arrays.asList(4L, 5L);

            if (lastPage.size() >= perPage) {
                //load the data
                //                List<Long> firstPage = Arrays.asList(1L, 2L, 3L);
                //                Long updatedMaxId = firstPage.get(firstPage.size() - 1);
                //recur
                //                theResponse = new DemoResponse(firstPage, paginatedQuery(updatedMaxId, perPage));
            } else {
                //last page
                theResponse = new DemoResponse(lastPage, Observable.<Response>empty());
            }

        }

        return Observable.just(theResponse);
    }

    public static interface Response {

        List<Long> data();

        Observable<Response> next();
    }

    public static class DemoResponse implements Response {

        private final List<Long> data;
        private final Observable<Response> next;

        DemoResponse(List<Long> data, Observable<Response> next) {

            this.data = data;
            this.next = next;
        }

        @Override
        public List<Long> data() {

            return data;
        }

        @Override
        public Observable<Response> next() {

            return next;
        }
    }