RxJS Test

RxJS.test.js

const Rx = require('rxjs/Rx');
const request = require('request');

//Convert node method to Rx observable
const requestGet$ = function(url) {
	let requestGetObservable = Rx.Observable.bindNodeCallback(request);
	return requestGetObservable(url)
		.do(x => console.log('requestGet$ is called'))
		.map(x => x[1])
		.map(x => JSON.parse(x))
		.catch(error => Rx.Observable.of(''));
}

requestGet$('http://ip.jsontest.com/')
	.subscribe(
		x => console.log(JSON.stringify(x.ip)),
		err => console.log(err)
	);

let ipCorrect$ = requestGet$('http://ip.jsontest.com/');
let ipInCorrect$ = requestGet$('http://dyabo.jsontest.com/');
let header$ = requestGet$('http://headers.jsontest.com/');

//Combine two observable results
Rx.Observable.combineLatest(ipCorrect$, header$, ipInCorrect$)
	.subscribe(ipAndHeader => {
		let [ip, header, ipInCorrect] = ipAndHeader;
		console.log(ip, header, ipInCorrect);
	});


let getAsyncValueFromCache = new Promise((resolve, reject) => {
	setTimeout(() => {
		let value = Math.random() * 10 > 5;
		console.log(`Value from getAsyncValueA is ${value}`);
		if (value) {
			reject(value)
		} else {
			resolve(value);
		}
	}, 1000);
});

let getAsyncValueFromCache$ = () => {
	return Rx.Observable.fromPromise(getAsyncValueFromCache).catch( e => Rx.Observable.throw('WE_HAVE_VALIDATION_VALUE_IN_CACHE') );
};


let calculateResult$ = Rx.Observable.interval(1000);

getAsyncValueFromCache$()
	.concatMap( resultFromCache => {
		if (!resultFromCache) {
			return calculateResult$;	
		}
	})
	.subscribe(
		validationResult => console.log(validationResult),
		error => {
			if (error === 'WE_HAVE_VALIDATION_VALUE_IN_CACHE') {
				//Return response
			} else {
				console.log(error);
			}
		}
	);