具有无限滚动的 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

您可以使用 Subject 并将您正在解决的问题分成 2 个可观察对象。一个用于滚动事件，另一个用于检索数据。例如：

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>

Answer 2

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

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

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

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

RxJs Observable with infinite scroll OR how to combine Observables

reactive-programming

observable

infinite-scroll

rxjs

问题：