RxJs Observed with infinite scrolling OR how to combine Observables

Question

I have tableone that uses infinite scrolling to load more results and add them when the user reaches the bottom of the page.

I currently have the following code:

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);
            }
        )
    }
}

There is one main problem :

Called every time appendNextPage, I get a completely new one Observable, which then calls all previous HTTP calls over and over again.

A minor issue is that this code is ugly, and it looks too much like such a simple use case.

Questions:

How to solve this problem in a beautiful way?

Observables -, ?

Answer 1

, , , . , . , - :

var nextPage = Rx.Observable.fromEvent(page, 'nextpage');

, , - , . :

var pageStream = nextPage
        //Always trigger the first page to load
        .startWith(0)
        //Load these pages asynchronously, but keep them in order
        .concatMap((_, pageNum) => $http(...), (_, page) => page.content)
        //One option of how to join the pages together
        .scan((pages, p) => {
          pages.push(p);
          return pages;
        }, []);

reset, , reset.

resetPages
          //Again always start with a fresh load
          .startWith(0)
          //Anytime there is a reset, restart the internal stream.
          .flatMapLatest(_ => 
            nextPage.startWith(0)
                    .concatMap((_, pageNum) => $http(...), 
                             (_, page) => page.content)
                    .scan((pages, p) => {
                      pages.push(p);
                      return pages;
                    }, [])
          )
          .subscribe(x => /*Render page content*/);

, , , , ,

Answer 2

, , 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>