rx + transducers

rx_transducers.js

const map = f => r => (m, i) => r(m, f(i));
const filter = p => r => (m, i) => (p(i) ? r(m, i) : m);
function conc(m, i) { return m.concat(i); }

[1, 2, 3].reduce(
  map(x => (console.log('map ' + x), x + x))(
  filter(x => (console.log('filter ' + x), x !== 4))(conc)),
  []);

class S {
  constructor(subs, evs) { this.subs = subs || []; this.evs = evs || [] }
  sub(su, m) { return new S(this.subs.concat({ su, m: this.evs.reduce(su, m) })); }
  pub(e) { return new S(this.subs.map(r => ({ su: r.su, m: r.su(r.m, e) })), this.evs.concat(e)); }
}

const s = (new S).sub((m, i) => m.concat(i + 1), []).sub((m, i) => m.concat(i * 2), []).pub(2).pub(5);

const ss = { s, pub(e) { return this.s = this.s.pub(e); }, sub(s, m) { return this.s = this.s.sub(s, m); } };

ss.sub(
  map(x => (console.log('map ' + x), x + x))(
  filter(x => (console.log('filter ' + x), x !== 4))(conc))
, []);

ss.pub(1); ss.pub(2); ss.pub(3); ss.s.subs[2].m