ross-nordstrom/rxjs-recursive-poll.js

## 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();
	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();