[Javascript] Broadcaster + Operator + Listener pattern -- 8. Pass a Done Symbol when an Async Function is Done

All good things come to an end. The concept of "done" plays a large part in our pattern and asynchronous code in general. We have to be able to handle when a broadcaster has completed its passing values so we can know what to do next.

import { curry } from "lodash";

const log = (v) => console.log(v);

const done = Symbol("done");

export let createInterval = curry((time, listener) => {
  let i = 0;
  let id = setInterval(() => {
    listener(i++);
  }, time);
  return () => {
    clearInterval(id);
  };
});

const forOf = curry((itertor, listener) => {
  const id = setTimeout(() => {
    for (let item of itertor) {
      listener(item);
    }
    listener(done);
  }, 0);
  return () => {
    clearTimeout(id);
  };
});

let zip = curry((broadcaster1, broadcaster2, listener) => {
  let cancelBoth;
  let buffer1 = [];
  let cancel1 = broadcaster1((value) => {
    buffer1.push(value);
    if (buffer2.length) {
      listener([buffer1.shift(), buffer2.shift()]);

      if (buffer1[0] === done || buffer2[0] === done) {
        listener(done);
        cancelBoth();
      }
    }
  });

  let buffer2 = [];
  let cancel2 = broadcaster2((value) => {
    buffer2.push(value);

    if (buffer1.length) {
      listener([buffer1.shift(), buffer2.shift()]);

      if (buffer1[0] === done || buffer2[0] === done) {
        listener(done);
        cancelBoth();
      }
    }
  });

  cancelBoth = () => {
    cancel1();
    cancel2();
  };

  return cancelBoth;
});

let zipNumberIntervewl = zip(forOf([1, 2, 3]), createInterval(1000));

zipNumberIntervewl(log);