MessagePagingSource.java

public class MessagePagingSource extends RxPagingSource<Integer, Message> {

    private MessageRepository messageRepository;

    // the initial load size for the first page may be different from the requested size
    private int initialLoadSize;

    public MessagePagingSource(MessageRepository messageRepository) {
        this.messageRepository = messageRepository;
    }

    @Override
    public Single<LoadResult<Integer, Message>> loadSingle(LoadParams<Integer> params) {
            // Start refresh at page 1 if undefined.
        Integer nextPageNumber = params.getKey() == null ? 1 : params.getKey();

        if (params.getKey() == null) {
            initialLoadSize = params.getLoadSize();
        }

        // work out the offset into the database to retrieve records from the page number,
        // allow for a different load size for the first page
        Integer offset;
        if (nextPageNumber == 2)
            offset = initialLoadSize;
        else
            offset = ((nextPageNumber - 1) * params.getLoadSize()) + (initialLoadSize - params.getLoadSize());

        return RxJavaBridge.toV2Single(messageRepository.getMessages(params.getLoadSize(), offset)
                .subscribeOn(Schedulers.io())
                .toList()
                .map( list ->
                    {
                        // assume that if a full page is not loaded, that means the end of the data
                        Integer nextKey = list.size() < params.getLoadSize() ? null : nextPageNumber + 1;
                        LoadResult<Integer, Message> loadResult = new LoadResult.Page(list, null, nextKey, 0, 0);
                        return loadResult;
                    })
                .onErrorReturn( e ->
                {
                    return new LoadResult.Error(e);
                }));
    }
}