反应式编程-RxJs Observable与无限滚动或如何组合Observable

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

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

Answer 2

您可以使用Subject并将您正在解决的问题分离为两个可观测值。一个用于滚动事件，另一个用于检索数据。例如：

let scrollingSubject = new Rx.Subject();
let dataSubject = new Rx.Subject();
//store the data that has been received back from server to check if a page has been
// received previously
let dataList = [];
scrollingSubject.subscribe(function(page) {
    dataSubject.onNext({
        pageNumber: page,
        pageData: [page + 10] // the data from the server
    });
});
dataSubject.subscribe(function(data) {
    console.log('Received data for page ' + data.pageNumber);
    dataList.push(data);
});
//scroll to page 1
scrollingSubject.onNext(1);
//scroll to page 2
scrollingSubject.onNext(2);
//scroll to page 3
scrollingSubject.onNext(3);

 <script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/4.1.0/rx.all.js"></script>