Reactive Programming introduction

by Arnaud Buchholz

HTML

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.5.0/Rx.min.js"></script>
<link rel="stylesheet" href="https://maxcdn.bootstrapcdn.com/bootstrap/3.3.7/css/bootstrap.min.css">
<table class="table table-striped">
  <thead>
    <tr>
      <th>Step \ Source</th>
    </tr>
  </thead>
  <tbody>
    <tr><td>0</td></tr>
    <tr><td>1</td></tr>
    <tr><td>2</td></tr>
    <tr><td>3</td></tr>
    <tr><td>4</td></tr>
    <tr><td>5</td></tr>
    <tr><td>6</td></tr>
    <tr><td>7</td></tr>
    <tr><td>8</td></tr>
    <tr><td>9</td></tr>
    <tr class="total">
      <td>Total</td>
    </tr>
  </tbody>
</table>
<hr>
<button class="btn" id="synchronous">Imperative</button>
<button class="btn" id="asynchronous">Reactive</button>

CSS

.total {
  font-weight: bold;
}

JavaScript

function log (value, rowIndex, colIndex) {
    var now = new Date(),
        row = document.querySelectorAll(".table tbody tr")[rowIndex];
    while (new Date() - now < 50); // Create blocking wait of 50 ms
    row.querySelectorAll("td")[colIndex].innerHTML = value;
}

var synchronousSource = [0, "1", "foo", "2", 3, "5", "bar", 7, "11", 13];

function addColumn(name) {
    var th = document.createElement("th"),
        headTr = document.querySelector(".table thead tr");
    th.appendChild(document.createTextNode(name));
    headTr.appendChild(th);
    [].slice.call(document.querySelectorAll(".table tbody tr"))
        .forEach(function (tr) {
            var td = document.createElement("td");
            tr.appendChild(td).innerHTML = "&nbsp;";
        });
    return headTr.querySelectorAll("th").length - 1;
}

function process (source, name) {
    var colIndex = addColumn(name);
    // Process source
    return source
        .map(function (value, rowIndex) {
            log(value, rowIndex, colIndex);
            return parseInt(value, 10);
        })
        .filter(function (value) {
            return !isNaN(value);
        })
        .reduce(function (result, value) {
            result.total += value;
            return result;
        }, {
            colIndex: colIndex,
            total: 0
        });
}

function showProcessResult(result) {
    log(result.total, 10, result.colIndex);
}

document.getElementById("synchronous").addEventListener("click", function () {
    showProcessResult(process(synchronousSource, "Imp"));
});

function toAsynchronous(source) {
    return Rx.Observable.interval(1).take(source.length)
        .map(function (index) {
            return source[index];
        });
}

document.getElementById("asynchronous").addEventListener("click", function () {
    process(toAsynchronous(synchronousSource), "Rx")
        .subscribe(showProcessResult);
});