使用RxJava延迟获取分页对象

Question

我几乎卖给了RxJava,它是Retrofit的完美伴侣,但我在迁移代码时遇到了一个共同的模式:为了节省带宽,我想懒得从我的webservice中获取(分页)对象,而我的listview(或recyclerview)使用反应式编程滚动.

我以前的代码完美地完成了这项工作,但是反应式编程似乎值得尝试.

听listview/recyclerview滚动(和其他无聊的东西)不是问题,使用Retrofit很容易获得Observable:

@GET("/api/messages")
Observable<List<Message>> getMessages(@Path("offset") int offset, @Path("limit") int limit);

我只是无法弄清楚在反应式编程中使用的模式.

该Concat操作似乎是一个很好的起点,随着ConnectableObservable在某个时刻推迟发射,也许flatMap,但如何？

编辑:

这是我目前的(幼稚)解决方案:

public interface Paged<T> {
    boolean isLoading();
    void cancel();
    void next(int count);
    void next(int count, Scheduler scheduler);
    Observable<List<T>> asObservable();
    boolean hasCompleted();
    int position();
}

我使用一个主题实现:

public abstract class SimplePaged<T> implements Paged<T> {

    final PublishSubject<List<T>> subject = PublishSubject.create();
    private volatile boolean loading;
    private volatile int offset;
    private Subscription subscription;

    @Override
    public boolean isLoading() {
        return loading;
    }

    @Override
    public synchronized void cancel() {
        if(subscription != null && !subscription.isUnsubscribed())
            subscription.unsubscribe();

        if(!hasCompleted())
            subject.onCompleted();

        subscription = null;
        loading = false;
    }

    @Override
    public void next(int count) {
        next(count, null);
    }

    @Override
    public synchronized void next(int count, Scheduler scheduler) {
        if (isLoading())
            throw new IllegalStateException("you can't call next() before onNext()");

        if(hasCompleted())
            throw new IllegalStateException("you can't call next() after onCompleted()");

        loading = true;

        Observable<List<T>> obs = onNextPage(offset, count).single();

        if(scheduler != null)
            obs = obs.subscribeOn(scheduler); // BEWARE! onNext/onError/onComplete will happens on that scheduler!

        subscription = obs.subscribe(this::onNext, this::onError, this::onComplete);
    }

    @Override
    public Observable<List<T>> asObservable() {
        return subject.asObservable();
    }

    @Override
    public boolean hasCompleted() {
        return subject.hasCompleted();
    }

    @Override
    public int position() {
        return offset;
    }

    /* Warning: functions below may be called from another thread */
    protected synchronized void onNext(List<T> items) {
        if (items != null)
            offset += items.size();

        loading = false;

        if (items == null || items.size() == 0)
            subject.onCompleted();
        else
            subject.onNext(items);
    }

    protected synchronized void onError(Throwable t) {
        loading = false;
        subject.onError(t);
    }

    protected synchronized void onComplete() {
        loading = false;
    }

    abstract protected Observable<List<T>> onNextPage(int offset, int count);

}

Answer 1

这是处理反应性分页的几种可能方法中的一种.假设我们有一个方法getNextPageTrigger,Observable当滚动侦听器(或任何输入)想要加载新页面时,它返回发出一些事件对象.在现实生活中它可能有debounce操作员,但除此之外我们将确保我们只在最新页面加载后触发它.

我们还定义了一种从列表中解包消息的方法:

Observable<Message> getPage(final int page) {
  return service.getMessages(page * PAGE_SIZE, PAGE_SIZE)
      .flatMap(messageList -> Observable.from(messageList));
}

然后我们可以制作实际的获取逻辑:

// Start with the first page.
getPage(0)
    // Add on each incremental future page.
    .concatWith(Observable.range(1, Integer.MAX_VALUE)
        // Uses a little trick to get the next page to wait for a signal to load.
        // By ignoring all actual elements emitted and casting, the trigger must
        // complete before the actual page request will be made.
        .concatMap(page -> getNextPageTrigger().limit(1)
            .ignoreElements()
            .cast(Message.class)
            .concatWith(getPage(page)))  // Then subscribe, etc..

这仍然缺少一些可能重要的事情:

1 - 这显然不知道何时停止获取其他页面,这意味着一旦它到达终点,取决于服务器返回的内容,它可能会在每次触发滚动时保持命中错误或清空结果.解决此问题的方法取决于您如何向客户端发出信号,表明不再需要加载页面.

2 - 如果您需要重试错误,我建议您查看retryWhen运营商.否则,常见的网络错误可能导致页面加载中的错误传播.