callbag-create

callbag-create.js

const create = prod => (start, sink) => {
  if (start !== 0) return;
  if (typeof prod !== 'function') {
    sink(0, () => {});
    return;
  }
  let init = got2 = false;
  let unsub;
  sink(0, t => {
    got2 = got2 || t === 2;
    got2 && unsub && unsub();
    if ((got2 && unsub) || init) return;
    unsub = prod((pt, pd) => {
      if (pt === 0 || got2) return;
      init = true;
      got2 = got2 || pt === 2;
      sink(pt, pd);
    });
  });
};

// Example 1:
const unsub = pipe(
  create((sink) => {
    let val = 0;
    const id = setInterval(() => sink(1, val++), 100);
    return () => clearInterval(id);
  }),
  map(n => n + 1),
  subscribe({
    next(n) {
      console.log('Val: ', n);
    }
  })
);

unsub(); // immediately unsubscribes, clearing the interval inside the producer. Nothing is logged.

// Example 2 (same code as above but unsub is called after 500ms):
setTimeout(unsub, 500);
// logs 1, 2, 3, 4 and clears the interval inside the producer.