Callbag-ish Range

range.ts

export function range(count: number, start = 0, step = 1): Stream<number> {
  return (snk) => {
    let open = true;
    let index = Math.floor(Math.max(0, count));
    let value = start;
    let readyCount = 0;
    let pulling = false;

    const close = () => open = false;
    const talk = snk((count) => {
      readyCount += count;
      pull();
    });

    function pull() {
      if (open && !pulling) {
        pulling = true;

        // Empty range while open and readyCount not empty
        while (open && readyCount > 0 && index > 0) {
          talk.event(value);
          value += step;
          readyCount--;
          index--;
        }

        // Reached end of range without closing, close and end
        if (open && readyCount > 0) {
          close();
          talk.end();
        }
        pulling = false;
      }
    }

    return disposable(close);
  };
}