👜 Callbag factory that combines data from multiple callbag sources
MIT License
/**
npm install callbag-combine
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;