可无限滚动的RxJs Observable或如何组合Observables

Question

当用户到达页面底部时，我table使用无限滚动滚动来加载更多结果并将其附加。

目前，我有以下代码：

var currentPage = 0;
var tableContent = Rx.Observable.empty();

function getHTTPDataPageObservable(pageNumber) {
    return Rx.Observable.fromPromise($http(...));
}

function init() {
    reset();
}

function reset() {
    currentPage = 0;
    tableContent = Rx.Observable.empty();
    appendNextPage();
}

function appendNextPage() {
    if(currentPage == 0) {
        tableContent = getHTTPDataPageObservable(++currentPage)
                .map(function(page) { return page.content; });
    } else {
        tableContent = tableContent.combineLatest(
            getHTTPDataPageObservable(++currentPage)
                    .map(function(page) { return page.content; }),
            function(o1, o2) {
                return o1.concat(o2);
            }
        )
    }
}

有一个主要问题：

每次appendNextPage调用时，我都会得到一个全新的信息Observable，它会一次又一次触发所有先前的HTTP调用。

一个小问题是，这段代码很难看，对于这样一个简单的用例而言，看起来太多了。

问题：

如何以一种很好的方式解决这个问题？

有可能以Observables不同的方式组合它们，而又不会一次又一次地触发整个堆栈吗？

Answer 1

您没有包含它，但我假设您有某种方法可以检测用户何时到达页面底部。可用于触发新加载的事件。为了这个答案，我会说您已将其定义为：

const nextPage = fromEvent(page, 'nextpage');

您真正想做的是尝试将其映射到一个定向流的流，而不是将流用作可变对象。因此：

const pageStream = nextPage.pipe(
  //Always trigger the first page to load
  startWith(0),

  //Load these pages asynchronously, but keep them in order
  concatMap(
    (_, pageNum) => from($http(...)).pipe(pluck('content')) 
  ),
        
  //One option of how to join the pages together
  scan((pages, p) => ([...pages, p]), [])
)

;

如果您需要重置功能，我建议您还考虑包装整个流以触发重置。

resetPages.pipe(
  // Used for the "first" reset when the page first loads
  startWith(0),

  //Anytime there is a reset, restart the internal stream.
  switchMapTo( 
    nextPage.pipe(
      startWith(0),
      concatMap(
        (_, pageNum) => from($http(...)).pipe(pluck('content'))
      ),
      scan((pages, p) => ([...pages, p]), [])
  )
).subscribe(x => /*Render page content*/);

如您所见，通过重构将逻辑嵌套到流中，我们可以删除之前浮动的全局状态