使用 RxJs 将分页请求转换为 Observable 流 [英] Turning paginated requests into an Observable stream with RxJs

问题描述

我有一个以页面为单位返回数据的服务.对一页的响应包含有关如何查询下一页的详细信息.

I have a service which returns data in pages. The response to one page contains details on how to query for the next page.

我的方法是返回响应数据，然后如果有更多可用页面，则立即将延迟调用连接到相同的可观察序列.

My approach is to return the response data and then immediately concat a deferred call to the same observable sequence if there are more pages available.

function getPageFromServer(index) {
  // return dummy data for testcase
  return {nextpage:index+1, data:[1,2,3]};
}

function getPagedItems(index) {
  return Observable.return(getPageFromServer(index))
    .flatMap(function(response) {
      if (response.nextpage !== null) {
        return Observable.fromArray(response.data)
          .concat(Observable.defer(function() {return getPagedItems(response.nextpage);}));
      }

      return Observable.fromArray(response.data);
    });
}

getPagedItems(0).subscribe(
  function(item) {
    console.log(new Date(), item);
  },
  function(error) {
    console.log(error);
  }
)

这一定是错误的方法，因为在 2 秒内你会得到:

This must be the wrong approach, because within 2 seconds you get:

      throw e;
            ^
RangeError: Maximum call stack size exceeded
    at CompositeDisposablePrototype.dispose (/Users/me/node_modules/rx/dist/rx.all.js:654:51)

正确的分页方法是什么?

What is the correct approach to pagination?

推荐答案

EDIT 啊！我看到你面临的问题.一些尾调用优化应该可以解决您的问题:

EDIT Ah! I see the problem you're facing. A bit of tail call optimization should fix you up:

function mockGetPageAjaxCall(index) {
  // return dummy data for testcase
  return Promise.resolve({nextpage:index+1, data:[1,2,3]});
}

function getPageFromServer(index) {
  return Observable.create(function(obs) {
    mockGetPageAjaxCall(index).then(function(page) {
      obs.onNext(page);
    }).catch(function(err) {
      obs.onError(err)
    }).finally(function() {
      obs.onCompleted();
    });
  });
}

function getPagedItems(index) {
    return Observable.create(function(obs) {
        // create a delegate to do the work
        var disposable = new SerialDisposable();
        var recur = function(index) {
            disposable.setDisposable(getPageFromServer(index).retry().subscribe(function(page) {
                obs.onNext(page.items);
                if(page.nextpage === null) {
                  obs.onCompleted();   
                }

                // call the delegate recursively
                recur(page.nextpage);
            }));
        };

        // call the delegate to start it
        recur(0);

        return disposable;
    });
}

getPagedItems(0).subscribe(
  function(item) {
    console.log(new Date(), item);
  },
  function(error) {
    console.log(error);
  }
)

这篇关于使用 RxJs 将分页请求转换为 Observable 流的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！