Reactive Programming

reactive-01-streams.js

// https://jsbin.com/naleno/1/edit

// array
var source = ['1', '1', 'foo', '2', '3', '5', 'bar', '8', '13'];
var result1 = source;

console.log(result1);

result2 = source
  .map(x => parseInt(x))
  .filter(x => !isNaN(x))
  .reduce((x, y) => x + y);

console.log(result2);


// event streams
source = Rx.Observable.interval(400).take(9)
  .map(i => ['1', '1', 'foo', '2', '3', '5', 'bar', '8', '13'][i]);
result1 = source;

result1.subscribe(x => console.log(x));

result2 = source
  .map(x => parseInt(x))
  .filter(x => !isNaN(x))
  .reduce((x, y) => x + y);

result2.subscribe(x => console.log(x));

reactive-02-click-events.js

// https://jsbin.com/qocaqu/2/edit

// <a href="#" class="button">BUTTON</a><h4>-</h4>

// element references
var button = document.querySelector('.button');
var label = document.querySelector('h4');

// event stream
var clickStream = Rx.Observable.fromEvent(button, 'click');

// double click stream
var doubleClickStream = clickStream
  .buffer(() => clickStream.debounce(250))
  .map(arr => arr.length)
  .filter(len => len === 2);

doubleClickStream.subscribe(event => {
  label.textContent = 'double click';
});

doubleClickStream
  .delay(1000)
  .subscribe(suggestion => {
    label.textContent = '-';
  });

reactive-03-dynamic-behavior.js

// https://jsbin.com/yikabo/3/edit

// It allows you to specify the dynamic behavior of a value completely at the time of declaration.

var a = 3;
var b = 10 * a;
console.log(b);

a = 4;
console.log(b);

// Reactive
var streamA = Rx.Observable.of(3, 4);
var streamB = streamA.map(a => 10 * a);

streamB.subscribe(b => console.log(b));

reactive-04-flat-map.js

// https://jsbin.com/wavoqu/3/edit

var requestStream = Rx.Observable.just('https://api.github.com/users');

var responseStream = requestStream
  // map would create so called metastreams
  // flatMap flattens these stream branches onto the main stream (kind of promise.then)
  .flatMap(requestUrl => Rx.Observable.fromPromise(jQuery.getJSON(requestUrl)));

responseStream.subscribe(response => {
	console.log(response);
});

reactive-05-render-dom.js

// https://jsbin.com/lugiya/3/edit

var refreshButton = document.querySelector('.refresh');

var requestStream = Rx.Observable.just('https://api.github.com/users');

var responseStream = requestStream
  // kind of promise.then
  .flatMap(requestUrl => Rx.Observable.fromPromise(jQuery.getJSON(requestUrl)));

var createSuggestionStream = function(responseStream) {
  return responseStream
    .map(userList => userList[Math.floor(Math.random() * userList.length)]);
};

var renderSuggestion = function(user, selector) {
  var element = document.querySelector(selector);

  var usernameElement = element.querySelector('.username');
  usernameElement.href = user.html_url;
  usernameElement.textContent = user.login;

  var imageElement = element.querySelector('img');
  imageElement.src = user.avatar_url;
};

var suggestion1Stream = createSuggestionStream(responseStream);
var suggestion2Stream = createSuggestionStream(responseStream);
var suggestion3Stream = createSuggestionStream(responseStream);

suggestion1Stream.subscribe(user => {
	renderSuggestion(user, '.suggestion1');
});

suggestion2Stream.subscribe(user => {
	renderSuggestion(user, '.suggestion2');
});

suggestion3Stream.subscribe(user => {
	renderSuggestion(user, '.suggestion3');
});

reactive-06-refresh.js

// https://jsbin.com/duzejo/6/edit

var refreshButton = document.querySelector('.refresh');

var refreshClickStream = Rx.Observable.fromEvent(refreshButton, 'click');

var startupRequestStream = Rx.Observable.just('https://api.github.com/users');

var requestOnRefreshStream = refreshClickStream
  .map(event => {
	  var randomOffset = Math.floor(Math.random() * 500);
	  return 'https://api.github.com/users?since=' + randomOffset;
  });

// merge returns events from either stream one or stream two
//
// ------a---b------c----->
// s---------------------->
//          merge
// s-----a---b------c----->
//
var responseStream = requestOnRefreshStream.merge(startupRequestStream)
  .flatMap(requestUrl => Rx.Observable.fromPromise(jQuery.getJSON(requestUrl)));

var createSuggestionStream = function(responseStream) {
  return responseStream
    .map(userList => userList[Math.floor(Math.random() * userList.length)]);
};

var renderSuggestion = function(user, selector) {
  var element = document.querySelector(selector);
  if (user === null) {
    element.style.visibility = 'hidden';
  } else {
    element.style.visibility = 'visible';
    var usernameElement = element.querySelector('.username');
    usernameElement.href = user.html_url;
    usernameElement.textContent = user.login;

    var imageElement = element.querySelector('img');
	imageElement.src = '';
    imageElement.src = user.avatar_url;
  }
};

var suggestion1Stream = createSuggestionStream(responseStream);
var suggestion2Stream = createSuggestionStream(responseStream);
var suggestion3Stream = createSuggestionStream(responseStream);

suggestion1Stream.subscribe(user => {
	renderSuggestion(user, '.suggestion1');
});

suggestion2Stream.subscribe(user => {
	renderSuggestion(user, '.suggestion2');
});

suggestion3Stream.subscribe(user => {
	renderSuggestion(user, '.suggestion3');
});

reactive-07-start-with.js

// https://jsbin.com/xusade/3/edit

var refreshButton = document.querySelector('.refresh');

var refreshClickStream = Rx.Observable.fromEvent(refreshButton, 'click');

var startupRequestStream = Rx.Observable.just('https://api.github.com/users');

var requestOnRefreshStream = refreshClickStream
  .map(event => {
	  var randomOffset = Math.floor(Math.random() * 500);
	  return 'https://api.github.com/users?since=' + randomOffset;
  });

// merge returns events from either stream one or stream two

// ------a---b------c----->
// s---------------------->
//          merge
// s-----a---b------c----->

var responseStream = startupRequestStream.merge(requestOnRefreshStream)
  .flatMap(requestUrl => Rx.Observable.fromPromise(jQuery.getJSON(requestUrl)));

// -----u----------u------>
//  startWith(N)
// N----u----------------->

// ---------N---N--------->
//       merge
// N----u---N---N--u------>

var createSuggestionStream = function(responseStream) {
  return responseStream
    .map(userList =>
		 userList[Math.floor(Math.random() * userList.length)]
	)
	.startWith(null)
    .merge(refreshClickStream.map(event => null));
};

var renderSuggestion = function(user, selector) {
  var element = document.querySelector(selector);
  if (user === null) {
    element.style.visibility = 'hidden';
  } else {
    element.style.visibility = 'visible';
    var usernameElement = element.querySelector('.username');
    usernameElement.href = user.html_url;
    usernameElement.textContent = user.login;

    var imageElement = element.querySelector('img');
	imageElement.src = '';
    imageElement.src = user.avatar_url;
  }
};

var suggestion1Stream = createSuggestionStream(responseStream);
var suggestion2Stream = createSuggestionStream(responseStream);
var suggestion3Stream = createSuggestionStream(responseStream);

suggestion1Stream.subscribe(user => {
	renderSuggestion(user, '.suggestion1');
});

suggestion2Stream.subscribe(user => {
	renderSuggestion(user, '.suggestion2');
});

suggestion3Stream.subscribe(user => {
	renderSuggestion(user, '.suggestion3');
});

reactive-08-share-streams.js

// https://jsbin.com/laleni/4/edit

var responseStream = startupRequestStream.merge(requestOnRefreshStream)
  .flatMap(requestUrl => {
	  console.log('do network request');
	  return Rx.Observable.fromPromise(jQuery.getJSON(requestUrl))
  })
  // share subscriptions for all subscribers
  .shareReplay(1);

reactive-09-cached.js

// https://jsbin.com/sugama/5/edit

var refreshButton = document.querySelector('.refresh');
var closeButton1 = document.querySelector('.close1');
var closeButton2 = document.querySelector('.close2');
var closeButton3 = document.querySelector('.close3');

var refreshClickStream = Rx.Observable.fromEvent(refreshButton, 'click');
var close1ClickStream = Rx.Observable.fromEvent(closeButton1, 'click');
var close2ClickStream = Rx.Observable.fromEvent(closeButton2, 'click');
var close3ClickStream = Rx.Observable.fromEvent(closeButton3, 'click');

var startupRequestStream = Rx.Observable.just('https://api.github.com/users');

var requestOnRefreshStream = refreshClickStream
  .map(event => {
	  var randomOffset = Math.floor(Math.random() * 500);
	  return 'https://api.github.com/users?since=' + randomOffset;
  });

var requestStream = startupRequestStream.merge(requestOnRefreshStream);

var responseStream = requestStream
  .flatMap(requestUrl => {
	  console.log('do network request');
	  return Rx.Observable.fromPromise(jQuery.getJSON(requestUrl))
  })
  .shareReplay(1);

// refreshClickStream: -------f------------>
// requestStream:      r------r------------>
// responseStream:     ---R------R--------->
// closeClickStream:   --------------x----->
// suggestion1Stream:  N--u---N--u---u----->

var getRandomUser = function(userList) {
  return userList[Math.floor(Math.random() * userList.length)];	
};

var createSuggestionStream = function(responseStream, closeClickStream) {	
  return responseStream
    .map(getRandomUser)
	.startWith(null)
    .merge(refreshClickStream.map(event => null))
    .merge(closeClickStream.withLatestFrom(responseStream, (event, userList) => getRandomUser(userList)));
};

var renderSuggestion = function(user, selector) {
  var element = document.querySelector(selector);
  if (user === null) {
    element.style.visibility = 'hidden';
  } else {
    element.style.visibility = 'visible';
    var usernameElement = element.querySelector('.username');
    usernameElement.href = user.html_url;
    usernameElement.textContent = user.login;

    var imageElement = element.querySelector('img');
	imageElement.src = '';
    imageElement.src = user.avatar_url;
  }
};

var suggestion1Stream = createSuggestionStream(responseStream, close1ClickStream);
var suggestion2Stream = createSuggestionStream(responseStream, close2ClickStream);
var suggestion3Stream = createSuggestionStream(responseStream, close3ClickStream);

suggestion1Stream.subscribe(user => {
	renderSuggestion(user, '.suggestion1');
});

suggestion2Stream.subscribe(user => {
	renderSuggestion(user, '.suggestion2');
});

suggestion3Stream.subscribe(user => {
	renderSuggestion(user, '.suggestion3');
});

reactive-10-switch-map-to.js

// https://jsbin.com/jimitik/2/edit

const startButton = document.querySelector('#start');
const startButtonClickStream = Rx.Observable.fromEvent(startButton, 'click');
const intervalStream = Rx.Observable.interval(1000);
const startIntervalStream = startButtonClickStream
    .switchMapTo(intervalStream);

startIntervalStream.subscribe(x => console.log(x));

reactive-11-take-until.js

// https://jsbin.com/jimitik/2/edit

const startButton = document.querySelector('#start');
const stopButton = document.querySelector('#stop');
	  
const startButtonClickStream = Rx.Observable.fromEvent(startButton, 'click');
const stopButtonClickStream = Rx.Observable.fromEvent(stopButton, 'click');

const intervalStream = Rx.Observable.interval(1000);
const intervalThatStopStream = intervalStream.takeUntil(stopButtonClickStream);

startButtonClickStream
  .switchMapTo(intervalThatStopStream)
  .subscribe(x => console.log(x));

reactive-12-scan.js

// https://jsbin.com/nifedo/1/edit

const startButton = document.querySelector('#start');
const stopButton = document.querySelector('#stop');
	  
const startButtonClickStream = Rx.Observable.fromEvent(startButton, 'click');
const stopButtonClickStream = Rx.Observable.fromEvent(stopButton, 'click');

const intervalStream = Rx.Observable.interval(1000);
const intervalThatStopStream = intervalStream.takeUntil(stopButtonClickStream);

startButtonClickStream
  .switchMapTo(intervalThatStopStream)
  .scan(acc => {
    return {count: acc.count + 1}
  }, {count : 0})
  .subscribe(x => console.log(x));

reactive-13-start-with.js

// https://jsbin.com/qojidi/1/edit

const startButton = document.querySelector('#start');
const stopButton = document.querySelector('#stop');
	  
const startButtonClickStream = Rx.Observable.fromEvent(startButton, 'click');
const stopButtonClickStream = Rx.Observable.fromEvent(stopButton, 'click');

const intervalStream = Rx.Observable.interval(1000);
const intervalThatStopStream = intervalStream.takeUntil(stopButtonClickStream);

const data = {count: 0};

startButtonClickStream
  .switchMapTo(intervalThatStopStream)
  .startWith(data)
  .scan(acc => {
    return {count: acc.count + 1}
  })
  .subscribe(x => console.log(x));

reactive-14-map-to.js

// https://jsbin.com/jedoga/1/edit

const startButton = document.querySelector('#start');
const stopButton = document.querySelector('#stop');
	  
const startButtonClickStream = Rx.Observable.fromEvent(startButton, 'click');
const stopButtonClickStream = Rx.Observable.fromEvent(stopButton, 'click');

const intervalStream = Rx.Observable.interval(1000);
const intervalThatStopStream = intervalStream.takeUntil(stopButtonClickStream);

const data = {count: 0};
const inc = (acc) => ({count: acc.count + 1});
const reset = (acc) => data;

startButtonClickStream
  .switchMapTo(intervalThatStopStream)
  .mapTo(inc)
//  .mapTo(reset)
  .startWith(data)
  .scan((acc, curr) => curr(acc))
  .subscribe(x => console.log(x));

reactive-15-merge.js

// https://jsbin.com/pugufe/1/edit

const startButton = document.querySelector('#start');
const stopButton = document.querySelector('#stop');
const resetButton = document.querySelector('#reset');
	  
const startButtonClickStream = Rx.Observable.fromEvent(startButton, 'click');
const stopButtonClickStream = Rx.Observable.fromEvent(stopButton, 'click');
const resetButtonClickStream = Rx.Observable.fromEvent(resetButton, 'click');

const intervalStream = Rx.Observable.interval(1000);
const intervalThatStopStream = intervalStream.takeUntil(stopButtonClickStream);

const data = {count: 0};
const inc = (acc) => ({count: acc.count + 1});
const reset = (acc) => data;

const incOrResetStream = Rx.Observable.merge(
	intervalThatStopStream.mapTo(inc),
	resetButtonClickStream.mapTo(reset)
);

startButtonClickStream
  .switchMapTo(incOrResetStream)
  .startWith(data)
  .scan((acc, curr) => curr(acc))
  .subscribe(x => console.log(x));

reactive-16-composable-stream.js

// https://jsbin.com/xaguhin/2/edit

const startButton = document.querySelector('#start');
const halfButton = document.querySelector('#half');
const quarterButton = document.querySelector('#quarter');
const stopButton = document.querySelector('#stop');
const resetButton = document.querySelector('#reset');
	  
const startButtonClickStream = Rx.Observable.fromEvent(startButton, 'click');
const halfButtonClickStream = Rx.Observable.fromEvent(halfButton, 'click');
const quarterButtonClickStream = Rx.Observable.fromEvent(quarterButton, 'click');
const stopButtonClickStream = Rx.Observable.fromEvent(stopButton, 'click');
const resetButtonClickStream = Rx.Observable.fromEvent(resetButton, 'click');

const intervalStream = Rx.Observable.interval(1000);
const intervalThatStopStream = intervalStream.takeUntil(stopButtonClickStream);

const data = {count: 0};
const inc = (acc) => ({count: acc.count + 1});
const reset = (acc) => data;

const incOrResetStream = Rx.Observable.merge(
	intervalThatStopStream.mapTo(inc),
	resetButtonClickStream.mapTo(reset)
);

const startersClickStream = Rx.Observable.merge(
	startButtonClickStream.mapTo(1000),
	halfButtonClickStream.mapTo(500),
	quarterButtonClickStream.mapTo(250)
);

const intervalActions = (time) => Rx.Observable.merge(
	Rx.Observable.interval(time)
	  .takeUntil(stopButtonClickStream)
	  .mapTo(inc),
	resetButtonClickStream.mapTo(reset)
);

startersClickStream
  .switchMap(intervalActions)
  .startWith(data)
  .scan((acc, curr) => curr(acc))
  .subscribe(x => console.log(x));

reactive-17-input.js

// https://jsbin.com/faqewom/3/edit

const input = document.querySelector('#input');
const inputStream = Rx.Observable.fromEvent(input, 'input')
  .map(event => event.target.value);

inputStream.subscribe(x => console.log(x));

reactive-18-combine-latest.js

// https://jsbin.com/zomajo/1/edit

const startButton = document.querySelector('#start');
const halfButton = document.querySelector('#half');
const quarterButton = document.querySelector('#quarter');
const stopButton = document.querySelector('#stop');
const resetButton = document.querySelector('#reset');
	  
const startButtonClickStream = Rx.Observable.fromEvent(startButton, 'click');
const halfButtonClickStream = Rx.Observable.fromEvent(halfButton, 'click');
const quarterButtonClickStream = Rx.Observable.fromEvent(quarterButton, 'click');
const stopButtonClickStream = Rx.Observable.fromEvent(stopButton, 'click');
const resetButtonClickStream = Rx.Observable.fromEvent(resetButton, 'click');

const input = document.querySelector('#input');
const inputStream = Rx.Observable.fromEvent(input, 'input')
  .map(event => event.target.value);

const intervalStream = Rx.Observable.interval(1000);
const intervalThatStopStream = intervalStream.takeUntil(stopButtonClickStream);

const data = {count: 0};
const inc = (acc) => ({count: acc.count + 1});
const reset = (acc) => data;

const incOrResetStream = Rx.Observable.merge(
	intervalThatStopStream.mapTo(inc),
	resetButtonClickStream.mapTo(reset)
);

const startersClickStream = Rx.Observable.merge(
	startButtonClickStream.mapTo(1000),
	halfButtonClickStream.mapTo(500),
	quarterButtonClickStream.mapTo(250)
);

const intervalActions = (time) => Rx.Observable.merge(
	Rx.Observable.interval(time)
	  .takeUntil(stopButtonClickStream)
	  .mapTo(inc),
	resetButtonClickStream.mapTo(reset)
);

const timerStream = startersClickStream
  .switchMap(intervalActions)
  .startWith(data)
  .scan((acc, curr) => curr(acc));

Rx.Observable.combineLatest(
	timerStream,
	inputStream,
	(timer, input) => ({count: timer.count, text: input})
).subscribe(
	x => console.log(x),
	e => console.log(e),
	() => console.log('complete')
);

reactive-19-filter.js

// https://jsbin.com/vunami/1/edit

Rx.Observable.combineLatest(
	timerStream,
	inputStream,
	(timer, input) => ({count: timer.count, text: input})
)
  .filter(data => data.count === parseInt(data.text))
  .subscribe(
	x => console.log(x),
	e => console.log(e),
    () => console.log('complete')
  );

reactive-20-take-while.js

// https://jsbin.com/vehetu/1/edit

Rx.Observable.combineLatest(
	timerStream,
	inputStream,
	(timer, input) => ({count: timer.count, text: input})
)
  .takeWhile(data => data.count <= 3)
  .filter(data => data.count === parseInt(data.text))
  .subscribe(
	x => console.log(x),
	e => console.log(e),
    () => console.log('complete')
  );

reactive-21-reduce.js

// https://jsbin.com/vorofe/1/edit

Rx.Observable.combineLatest(
	timerStream,
	inputStream,
	(timer, input) => ({count: timer.count, text: input})
)
  .takeWhile(data => data.count <= 3)
  .filter(data => data.count === parseInt(data.text))
  .reduce((acc, curr) => acc + 1, 0)
  .subscribe(
	x => console.log(x),
	e => console.log(e),
    () => console.log('complete')
  );

reactive-22-do.js

// https://jsbin.com/qoxaloz/1/edit

Rx.Observable.combineLatest(
	timerStream.do(x => console.log(x)),
	inputStream.do(x => console.log(x)),
	(timer, input) => ({count: timer.count, text: input})
)
  .takeWhile(data => data.count <= 3)
  .filter(data => data.count === parseInt(data.text))
  .reduce((acc, curr) => acc + 1, 0)
  .subscribe(
	x => console.log(x),
	e => console.log(e),
    () => console.log('complete')
  );

reactive-23-with-latest-from.js

// https://jsbin.com/bohuru/1/edit

timerStream
  .do(x => console.log(x))
  .takeWhile(data => data.count <= 3)
  .withLatestFrom(
	 inputStream.do(x => console.log(x)),
	 (timer, input) => ({count: timer.count, text: input})
  )
  .filter(data => data.count === parseInt(data.text))
  .reduce((acc, curr) => acc + 1, 0)
  .subscribe(
	x => console.log(x),
	e => console.log(e),
	() => console.log('complete')
  );

reactive-24-repeat.js

// https://jsbin.com/biqahi/1/edit

timerStream
  .do(x => console.log(x))
  .takeWhile(data => data.count <= 3)
  .withLatestFrom(
	 inputStream.do(x => console.log(x)),
	 (timer, input) => ({count: timer.count, text: input})
  )
  .filter(data => data.count === parseInt(data.text))
  .reduce((acc, curr) => acc + 1, 0)
  .repeat()
  .subscribe(
	x => console.log(x),
	e => console.log(e),
	() => console.log('complete')
  );

reactive-25-share.js

// https://jsbin.com/daleyus/1/edit

const runningGameStream = timerStream
  .do(x => console.log(x))
  .takeWhile(data => data.count <= 3)
  .withLatestFrom(
	 inputStream.do(x => console.log(x)),
	 (timer, input) => ({count: timer.count, text: input})
  )
  .share();

runningGameStream
  .repeat()
  .subscribe(() => input.value = '');

runningGameStream
  .filter(data => data.count === parseInt(data.text))
  .reduce((acc, curr) => acc + 1, 0)
  .repeat()
  .subscribe(
	x => console.log(x),
	e => console.log(e),
	() => console.log('complete')
  );