reactive-programming - 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*/);

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