Пагинация списков в Android с RxJava. Часть I

Отличная статья, как раз столкнулся с задачей пагинации с использованием rx.

Неплохо! Есть всякие мелочи по коду, но в целом норм :)

Правда, offset, конечно, редко применяют в виде количества загруженных элементов т.к. пока пользователь смотрит текущий стейт данные могут добавиться/удалиться и будут проблемы (неотрисовка части данных или дубликаты), но это уже оффтопик.

Хорошая статья, приятно читается и написана хорошо, спасибо вам.

Хотел немного порассуджать про Subject. По моему мнению они выглядят как костыль над самими потоками. Потому что под источником подразумевается обычно замкнутая система, которая сама решает, как порождать ей элементы и когда это делать. И, как и любое костыль над контрактом, сабджекты провоцируют людей чаще их (сабджектов) использовать и само их использование иногда ведет к нарушению идеологии реактивных потоков:) Безусловно, бывает, что их применение оправдано. Обычно это делается в том случае, когда по-другому источник невозможно создать или в целях оптимизации.

Когда мы говорили о задаче, я подразумевал, что мы будем создавать свой Observable, который будет эммитить объекты самостоятельно, а не делать это через сабджекты.

Например, он может эммитить дату последнего айтема, чтобы загружать с этой даты новые, или adapter.getItemCount(), чтобы эммитить сразу оффсет, по которому идти, как у вас и сделано.
Что-то типа:

public final class ScrollObservable {
    public static Observable<Integer> from(final RecyclerView rv) {
        return Observable.create(subscriber -> {
            final RecyclerView.OnScrollListener sl = new RecyclerView.OnScrollListener() {
                @Override
                public void onScrolled(RecyclerView recyclerView, int dx, int dy) {
                    if (!subscriber.isUnsubscribed()) {
                        final int position = getLastVisibleItemPosition();
                        final int limit = getLimit();
                        final int updatePosition = rv.getAdapter().getItemCount() - 1 - (limit / 2);
                        if (position >= updatePosition) {
                            subscriber.onNext(rv.getAdapter().getItemCount());
                        }
                    }
                }
            };
            rv.addOnScrollListener(sl);
            subscriber.add(Subscriptions.create(() -> rv.removeOnScrollListener(sl)));
        });
    }
}

В этом случае получается Обсервабл, который начинает бешено эммитить айтемы при достижении определеного порога. Чтобы это происходило всего один раз, нам не надо делать подписки и отписки от сабджектов или что-то еще, просто применяем distinctUntilChanged и получаем новые оффсеты только раз:

final Observable<Integer> offsetRequestObs = ScrollObservable.from(recyclerView).distinctUntilChanged();

Соответственно в onNext мы не будем запускать новый Observable, потому что это тоже не очень круто, тк они получаются оторваны от жизненного цикла родителя. Просто делаем switchMap и потом перенаправляем на ui, где обрабатываем:

offsetRequestObs.switchMap(offset -> getLoadingObservable(offset))
                .subscribeOn(Schedulers.from(BackgroundExecutor.getSafeBackgroundExecutor()))
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(loadNewItemsSubscriber);

В целом это решение делает то же самое, но, как мне кажется, оно лучше поддается расширению и это полноценный Observable, с котором можно делать все, что угодно. Кроме того, имеем связанный жизненный цикл потоков, что тоже важно.

Как я и говорил выше, сабджекты обычно зло и без них очень часто можно сделать лучше и лаконичнее. Не понимаю, почему их все часто используют. Кроме того, важным фактом тут является еще то, что мы не делаем нового наследника RecyclerView, но и в вашем примере можно добиться того же.