readme.js

May 2, 2018 · View on GitHub

/**

callbag-combine
Callbag factory that combines the latest data points from multiple (2 or
more) callbag sources. It delivers those latest values as an array. Works
with both pullable and listenable sources.
npm install callbag-combine
Example:

const interval = require('callbag-interval');

const observe = require('callbag-observe');

const combine = require('callbag-combine');

const source = combine(interval(100), interval(350));

observe(x => console.log(x))(source); // [2,0]

                                      // [3,0]

                                      // [4,0]

                                      // [5,0]

                                      // [6,0]

                                      // [6,1]

                                      // [7,1]

                                      // [8,1]

                                      // ...

*/

const EMPTY = {};

const combine = (...sources) => (start, sink) => { if (start !== 0) return; const n = sources.length; if (n === 0) { sink(0, () => {}); sink(1, []); sink(2); return; } let Ns = n; // start counter let Nd = n; // data counter let Ne = n; // end counter const vals = new Array(n); const sourceTalkbacks = new Array(n); const talkback = (t, d) => { if (t === 0) return; for (let i = 0; i < n; i++) sourceTalkbacks[i](t, d); }; sources.forEach((source, i) => { vals[i] = EMPTY; source(0, (t, d) => { if (t === 0) { sourceTalkbacks[i] = d; if (--Ns === 0) sink(0, talkback); } else if (t === 1) { const _Nd = !Nd ? 0 : vals[i] === EMPTY ? --Nd : Nd; vals[i] = d; if (_Nd === 0) { const arr = new Array(n); for (let j = 0; j < n; ++j) arr[j] = vals[j]; sink(1, arr); } } else if (t === 2) { if (--Ne === 0) sink(2); } else { sink(t, d); } }); }); };

module.exports = combine;