playing with async paging

.babelrc

{
  "presets": ["es2015"]
}

.gitignore

node_modules/
package-lock.json

package.json

{
  "devDependencies": {
    "babel-cli": "^6.26.0",
    "babel-preset-es2015": "^6.24.1"
  },
  "dependencies": {
    "rxjs": "^5.5.6"
  }
}

rx.js

import { Observable, Subject, ReplaySubject } from 'rxjs/Rx';

/* to run:
 *  $ npm install
 *  $ npx babel-node rx.js
 */

function makePagedFetcher() {
    var accum = [];
    var offset;

    const output = new ReplaySubject(1);

    var observer;

    Observable.create(obs => {
        observer = obs;
    }).concatMap(cnt => {
        const start = (offset || 0) + 1;
        // dummy array for testing
        const arr = new Array(cnt).fill().map((_,i) => start + i).filter(i => i < 15);
        return Observable.of(arr).map(arr => {
            const len = arr.length;
            if (len < cnt) {
                observer.complete();
            }
            offset = arr[len-1];
            accum = [...accum, ...arr];
            return accum;
        });
    }).subscribe(output);

    output.fetchPage = function(cnt=10) {
        observer.next(cnt);
    };

    return output;
}

const fetcher = makePagedFetcher();

fetcher.subscribe(x => console.log("a", x), console.err, _ => console.log("complete"))
fetcher.fetchPage(5)
fetcher.subscribe(x => console.log("b", x))
fetcher.fetchPage(5)
fetcher.fetchPage(20)
fetcher.subscribe(x => console.log("c", x), console.err, _ => console.log("c complete"))