Distpatches -- MVP signals.

dispatches.js

// Reduce any arraylike object.
function reduceIndexed(indexed, next, initial) {
  var accumulated = initial;

  for (var i = 0; i < indexed.length; i += 1)
    accumulated = next(accumulated, indexed[i]);

  return accumulated;
}

function callWith(v, f) {
  f(v);
  return v;
}

// Call every function in an array (or arraylike) with value.
// Returns value.
function dispatch(wire, value) {
  return reduceIndexed(wire, callWith, value);
}

function applyWith(a, f) {
  f.apply(null, a);
}

// Call every function in an array (or arraylike) with an array of values.
function dispatchN(wire, values) {
  return reduceIndexed(wire, applyWith, values);
}

// Add a value to an array, returning array.
function add(a, v) {
  a.push(v);
  return a;
}

// Remove value from array, returning array.
function remove(a, v) {
  var i = a.indexOf(v);
  if (i !== -1) a.splice(i, 1);
  return a;
}

// Just add with reverse argument order and return of value,
// for convenient reduction.
function addTo(v, a) {
  add(a, v);
  return v;
}

// Creates a new wire `b`. Whenever a new value is dispatched to `a`, callback
// `f` will be invoked with `b` and value `v`, giving `f` a chance to `dispatch`
// value to `b`.
//
// Note that wires have direction. Updates to a will be forwarded to b, but not
// the other way.
//
// @TODO I could add an "up the chain" backchannel where any `end` value
// signaled from `b` gets sent to `a`. This is essentially what accumulators
// do.
function republish(a, f) {
  var b = [];
  add(a, function listenA(v) {
    f(b, v);
  });
  return b;
}

function filter(wire, predicate) {
  return republish(wire, function republishFilter(b, v) {
    if (predicate(v)) dispatch(b, v);
  });
}

function map(wire, x) {
  return republish(wire, function republishMap(b, v) {
    dispatch(b, x(v));
  });
}

// Given wire, return a new wire that will dispatch values reduced with `next`.
function reductions(wire, next, initial) {
  // Accumulated state is kept track of via closure and is dispatched to `n`
  // listeners when updated.
  var accumulated = initial;
  return republish(wire, function republishReductions(b, v) {
    accumulated = next(accumulated, v);
    dispatch(b, accumulated);
  });
}

// Given an array of wires, "flatten" them into a single wire.
function combine(/* wires */) {
  var c = [];
  function republishToC(v) {
    dispatch(c, v);
  }
  reduceIndexed(arguments, addTo, republishToC);
  return c;
}

function asserts(wire, assert) {
  var prev = null;
  return republish(wire, function republishAsserts(b, curr) {
    if (assert(prev, curr)) dispatch(b, curr);
    prev = curr;
  });
}

function id(thing) {
  return thing;
}

function sample(wire, trigger, assemble) {
  assemble = assemble || id;
  var sampled = null;

  add(wire, function (value) {
    // Update sampled value when `wire` is updated.
    sampled = value;
  });

  return republish(trigger, function (b, v) {
    dispatch(b, assemble(sampled, v));
  });
}

// Given a wire, create a new wire that represents the previous value of
// `wire` whenever `wire` is updated.
function previously(wire) {
  var prev = null;
  return republish(wire, function (b, curr) {
    dispatch(b, prev);
    prev = curr;
  });
}

function print(wire) {
  add(wire, console.log);
}

function on(element, name, useCapture) {
  // @TODO could create backchannel by either returning 2 wires (one for
  // lifecycle events) or by creating methods on `wire` (start/stop), or by
  // dispatching signals to wire.
  //
  // The prob/advantage with dispatching signals is that the message is
  // forwarded to all 
  var wire = [];

  element.addEventListener(name, function (event) {
    dispatch(wire, event);
  }, !!useCapture);
  return wire;
}