pagination_using_rxjava.kt

/**
 * initialize all resources
 * set current page to 1
 * create paginator and subscribe to events
 */
override fun initialize() {
    currentPage = 1                                         // set page = 1
    paginator = PublishProcessor.create()                   // create PublishProcessor

    val d = paginator.onBackpressureDrop()
            .filter { !loading }                            // return if it is still loading
            .doOnNext { loading = view.showProgress() }     // loading = true
            .concatMap { contract.getUsersFromServer(it) }  // API call
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe({
                loading = view.hideProgress()               // loading = false
                view.showItems(it)                          // show items
                currentPage++                               // increment page
            }, {
                loading = view.hideProgress()               // loading = false
                view.showError(it.localizedMessage)         // show error
            })

    disposables.add(d)

    onLoadMore(currentPage)
}

/**
 * called when list is scrolled to its bottom
 * @param page current page (not used)
 */
override fun onLoadMore(page: Int) {
    paginator.onNext(currentPage)                           // increment page if not loading
}