RxJS: concat vs. forkJoin

<!DOCTYPE html>
<html>

  <head>
    <link rel="stylesheet" href="style.css">
  </head>

  <body>
    <h1>RxJS: concat vs. forkJoin</h1>
    
    <div id="results"></div>

    <script src="http://cdnjs.cloudflare.com/ajax/libs/rxjs/4.0.6/rx.all.js"></script>
    <script src="script.js"></script>
  </body>

</html>

// https://github.com/Reactive-Extensions/RxJS/

// Generic code to display resuts

var results = document.getElementById('results');
function showHTML(html)
{
  results.insertAdjacentHTML('beforeend', html);
}
function show(text)
{
  showHTML("<p>" + text + "<p>");
}
function showObject(obj)
{
  show("<p>" + JSON.stringify(obj) + "<p>");
}

// Make a promise out of a timeout - https://developer.mozilla.org/en/docs/Web/JavaScript/Reference/Global_Objects/Promise

var reference = Date.now();

function promiseInTime(timeout, value, willReject)
{
  return new Promise(
    // The resolver function is called with the ability to resolve or reject the promise
    function(resolve, reject) 
    {
      window.setTimeout(
        function afterTimeout() 
        {
          value.promiseTimeout = Date.now() - reference;
          if (willReject)
          {
            // No luck
            reject(value);
          }
          else
          {
            // We fulfill the promise!
            resolve(value);
          }
        }, timeout);
    }
  );
}

// The real code

function concatObservables()
{
  showHTML("<h2>concat of observables</h2>");

  var o1 = Rx.Observable.interval(1500).take(1).map(function (i) { return { n: i, id: 'First', ts: Date.now() - reference }});
  var o2 = Rx.Observable.interval(1000).take(2).map(function (i) { return { n: i, id: 'Second', ts: Date.now() - reference }});
  var o3 = Rx.Observable.interval(2000).take(1).map(function (i) { return { n: i, id: 'Third', ts: Date.now() - reference }});
  // Alternative
  var oa = Rx.Observable.timer(1200).map(function (i) { return { n: i, id: 'Alternative', ts: Date.now() - reference }});

  Rx.Observable.concat(o1, o2, oa, o3).subscribe(
      function onNext(v) { v.timestamp = Date.now() - reference; showObject(v); },
      function onError(e) { var ts = Date.now() - reference; showHTML("Error " + JSON.stringify(e) + " at " + ts); },
      function onCompleted() { var ts = Date.now() - reference; showHTML("Completed at " + ts); }
  );
}

function concatPromises(withFailure)
{
  showHTML("<h2>concat of promises</h2>");

  var p1 = promiseInTime(1500, { id: 'First'});
  var p2 = promiseInTime(1000, { id: 'Second' });
  var p3 = promiseInTime(2000, { id: 'Third' });
  var pa = promiseInTime(1200, { id: 'Failed? ' + !!withFailure }, withFailure);

  Rx.Observable.concat(p1, p2, pa, p3).subscribe(
      function onNext(v) { v.timestamp = Date.now() - reference; showObject(v); },
      function onError(e) { var ts = Date.now() - reference; showHTML("Error " + JSON.stringify(e) + " at " + ts); },
      function onCompleted() { var ts = Date.now() - reference; showHTML("Completed at " + ts); }
  );
}

function baseForkJoinObservables()
{
  showHTML("<h2>Base (functionless) forkJoin of observables</h2>");

  var o1 = Rx.Observable.interval(1500).take(1).map(function (i) { return { n: i, id: 'First', ts: Date.now() - reference }});
  var o2 = Rx.Observable.interval(1000).take(2).map(function (i) { return { n: i, id: 'Second', ts: Date.now() - reference }});
  var o3 = Rx.Observable.interval(2000).take(1).map(function (i) { return { n: i, id: 'Third', ts: Date.now() - reference }});
  // Alternative
  var oa = Rx.Observable.timer(1200).map(function (i) { return { n: i, id: 'Alternative', ts: Date.now() - reference }});

  Rx.Observable.forkJoin(o1, o2, oa, o3).subscribe(
      function onNext(v) { showHTML("<p>Next " + JSON.stringify(v) + "</p>"); },
      function onError(e) { showHTML("<p>Error " + JSON.stringify(e) + "</p>"); },
      function onCompleted() { showHTML("<p>Completed</p>"); }
  );
}

function forkJoinObservables()
{
  showHTML("<h2>forkJoin of observables</h2>");

  var o1 = Rx.Observable.interval(1500).take(1).map(function (i) { return { n: i, id: 'First', ts: Date.now() - reference }});
  var o2 = Rx.Observable.interval(1000).take(2).map(function (i) { return { n: i, id: 'Second', ts: Date.now() - reference }});
  var o3 = Rx.Observable.interval(2000).take(1).map(function (i) { return { n: i, id: 'Third', ts: Date.now() - reference }});
  // Alternative
  var oa = Rx.Observable.timer(1200).map(function (i) { return { n: i, id: 'Alternative', ts: Date.now() - reference }});

  Rx.Observable.forkJoin(o1, o2, oa, o3, function fj(o1, o2, oa, o3)
  {
    return { o1: o1, o2: o2, o3: o3, oa: oa };
  }).subscribe(
      function onNext(v) { showHTML("<p>Next " + JSON.stringify(v) + "</p>"); },
      function onError(e) { showHTML("<p>Error " + JSON.stringify(e) + "</p>"); },
      function onCompleted() { showHTML("<p>Completed</p>"); }
  );
}

function baseForkJoinPromises(withFailure)
{
  showHTML("<h2>Base (functionless) forkJoin of promises</h2>");

  var p1 = promiseInTime(1500, { id: 'First'});
  var p2 = promiseInTime(1000, { id: 'Second' });
  var p3 = promiseInTime(2000, { id: 'Third' });
  var pa = promiseInTime(1200, { id: 'Failed? ' + !!withFailure }, withFailure);

  Rx.Observable.forkJoin(p1, p2, pa, p3).subscribe(
      function onNext(v) { showHTML("<p>Next " + JSON.stringify(v) + "</p>"); },
      function onError(e) { showHTML("<p>Error " + JSON.stringify(e) + "</p>"); },
      function onCompleted() { showHTML("<p>Completed</p>"); }
  );
}

function forkJoinPromises(withFailure)
{
  showHTML("<h2>forkJoin of promises</h2>");

  var p1 = promiseInTime(1500, { id: 'First'});
  var p2 = promiseInTime(1000, { id: 'Second' });
  var p3 = promiseInTime(2000, { id: 'Third' });
  var pa = promiseInTime(1200, { id: 'Failed? ' + !!withFailure }, withFailure);

  Rx.Observable.forkJoin(p1, p2, pa, p3, function fj(p1, p2, pa, p3)
  {
    return { p1: p1, p2: p2, p3: p3, pa: pa };
  }).subscribe(
      function onNext(v) { showHTML("<p>Next " + JSON.stringify(v) + "</p>"); },
      function onError(e) { showHTML("<p>Error " + JSON.stringify(e) + "</p>"); },
      function onCompleted() { showHTML("<p>Completed</p>"); }
  );
}

// Ctrl+/ to comment / uncomment a line
// concatObservables();
// concatPromises(false);
// concatPromises(true);
// baseForkJoinObservables();
// forkJoinObservables();
// baseForkJoinPromises(false);
// baseForkJoinPromises(true);
// forkJoinPromises(false);
// forkJoinPromises(true);

#dragTarget
{
    background-image: url(https://raw.githubusercontent.com/Reactive-Extensions/RxJS/master/examples/dragndrop/logo.png);
    background-repeat: no-repeat;
    background-position: center;
    background-size: contain;
    height: 200px;
    width: 200px;
    background-color: #000000;
    border: 1px solid #666666;
    color: #ffffff;
    padding: 10px;
    position: absolute;
    cursor: move;
}

# RxJS: concat vs. forkJoin

Trying Reactive Extensions for JavaScript.

Playground to deepen knowledge of operators.