Demonstration of using RxJS to implement a recursive API Poller. This is an improvement on the simple `Observable.interval()` approach, because we won't kill long-running queries.

rxjs-recursive-poll.js

const rxjs = require('rxjs');

/**
 * The entrypoint function. Pass it the thing to call
 */
function dynamicPoll(pollFunc/*, delayFunc*/) {
    // A recursive stream, which triggers   an API call
    let i = 0; // just here for debugging
    const trigger = new rxjs.BehaviorSubject(i++);

    // The polling observable.
    // While subscribed-to, it will continually call the "API", then queue up the next call
    // Simple extensions:
    //  - Delay the recursive API call (perhaps dynamic delay based on the previous call's latency)
    //  - Use closure state to implement incremental queries (e.g. maintain and use a 'modifiedSince' option)
    return trigger.concatMap(() => pollFunc(/*query, options*/)).map(res => {
        trigger.next(i++); // RECURSE!
        return res;
    });
}

/**
 * Mock API call
 * Uses a delay to emulate an API with variable latency
 * @return { time:string, delay:number, records:string[] }
 */
function fakeApi(/*query, options*/) {
    const delay = random(250, 10*1000); // random delay between 250ms - 10s
    const fakeRes = { time: shortDate(), delay: delay, records: ['fake api response'] };

    return rxjs.Observable.of(fakeRes).delay(delay);
}

function random(min, max) {
    return Math.round(Math.random() * (max - min) + min);
}

function shortDate() {
    return new Date().toTimeString().slice(0,8); // e.g. "11:35:13"
}

function demo() {
    console.warn('START DEMO');
    let myDemo = dynamicPoll(fakeApi).subscribe(res => console.info('API Res: ' + JSON.stringify(res)));

    setTimeout(() => {
        myDemo.unsubscribe();
        console.warn('END DEMO');
    }, 60*1000);
}

// TO RUN, SIMPLY:
// demo();

Example output: (just copy/paste the Gist into a node console

$ node --harmony

> demo()
START DEMO
undefined
> API Res: {"time":"11:48:37","delay":7186,"records":["fake api response"]}
API Res: {"time":"11:48:45","delay":1854,"records":["fake api response"]}
API Res: {"time":"11:48:46","delay":701,"records":["fake api response"]}
API Res: {"time":"11:48:47","delay":7534,"records":["fake api response"]}
API Res: {"time":"11:48:55","delay":5279,"records":["fake api response"]}
API Res: {"time":"11:49:00","delay":2450,"records":["fake api response"]}
API Res: {"time":"11:49:02","delay":1294,"records":["fake api response"]}
API Res: {"time":"11:49:04","delay":2938,"records":["fake api response"]}
API Res: {"time":"11:49:07","delay":1224,"records":["fake api response"]}
API Res: {"time":"11:49:08","delay":4489,"records":["fake api response"]}
API Res: {"time":"11:49:12","delay":3726,"records":["fake api response"]}
API Res: {"time":"11:49:16","delay":3944,"records":["fake api response"]}
API Res: {"time":"11:49:20","delay":1489,"records":["fake api response"]}
API Res: {"time":"11:49:21","delay":9950,"records":["fake api response"]}
END DEMO