javascript - 使用 RxJs 将分页请求转换为 Observable 流

Question

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

我的方法是返回响应数据，然后在有更多可用页面时立即连接对相同可观察序列的延迟调用。

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 秒内你会得到:

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

什么是正确的分页方法？

Answer 1

看了OP代码，果然是对的方法。只需要使您的模拟服务异步即可模拟真实情况。这是一个避免堆栈耗尽的解决方案(我还让 getPageFromServer 实际上返回一个冷的可观察对象，而不是要求调用者包装它)。

请注意，如果您真的希望您的服务请求在实际应用程序中同步完成，因此您需要确保您的代码在发生这种情况时不会耗尽堆栈，只需让 getPagedItems() 调用currentThread scheduler . currentThread 调度程序使用蹦床来调度任务，以防止递归调用(和堆栈耗尽)。查看 getPagedItems

末尾的注释行

function getPageFromServer(index) {
    // return dummy data asynchronously for testcase
    // use timeout scheduler to force the result to be asynchronous like
    // it would be for a real service request
    return Rx.Observable.return({nextpage: index + 1, data: [1,2,3]}, Rx.Scheduler.timeout);

    // for real request, if you are using jQuery, just use rxjs-jquery and return:
    //return Rx.Observable.defer(function () { return $.ajaxAsObservable(...); });
}

function getPagedItems(index) {
    var result = getPageFromServer(index)
        .retry(3) // retry the call if it fails
        .flatMap(function (response) {
            var result = Rx.Observable.fromArray(response.data);
            if (response.nextpage !== null) {
                result = result.concat(getPagedItems(response.nextpage));
            }
            return result;
        });

    // If you think you will really satisfy requests synchronously, then instead
    // of using the Rx.Scheduler.timeout in getPageFromServer(), you can
    // use the currentThreadScheduler here to prevent the stack exhaustion...

    // return result.observeOn(Rx.Scheduler.currentThread) 
    return result;
}