[Javascript] Broadcaster + Operator + Listener pattern -- 15. Create a Sequence of Broadcasters

import { addListener, createTimeout, done } from "./broadcasters";
import { hardCode } from "./operators";

const log = console.log;

const delayMessage = (message) => hardCode(message)(createTimeout(1000));

const sequence = (...broadcasters) => (listener) => {
  let broadcaster = broadcasters.shift();
  let cancel;
  let sequenceListener = (value) => {
    if (value === done && broadcasters.length) {
      broadcaster = broadcasters.shift();
      cancel = broadcaster(sequenceListener);
      return;
    }
    listener(value);
  };
  cancel = broadcaster(sequenceListener);

  return () => {
    cancel();
  };
};

let cancelMsgs = sequence(
  delayMessage("Hello"),
  delayMessage("I"),
  delayMessage("am"),
  delayMessage("Zhentian"),
  delayMessage("Wan")
)(log);

let inputClick = addListener("#input", "click");
inputClick(cancelMsgs);


// broadcasters.js

import { curry } from "lodash";
export let done = Symbol("done");

export let createTimeout = curry((time, listener) => {
  let id = setTimeout(() => {
    listener(null);
    listener(done);
  }, time);

  return () => {
    clearTimeout(id);
  };
});

export let addListener = curry((selector, eventType, listener) => {
  let element = document.querySelector(selector);
  element.addEventListener(eventType, listener);

  return () => {
    element.removeEventListener(eventType, listener);
  };
});

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

//broadcaster = function that accepts a listener
export let merge = curry((broadcaster1, broadcaster2, listener) => {
  let cancel1 = broadcaster1(listener);
  let cancel2 = broadcaster2(listener);

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

export let zip = curry((broadcaster1, broadcaster2, listener) => {
  let cancelBoth;

  let buffer1 = [];
  let cancel1 = broadcaster1((value) => {
    buffer1.push(value);
    // console.log(buffer1)
    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;
});

export let forOf = curry((iterable, listener) => {
  let id = setTimeout(() => {
    for (let i of iterable) {
      listener(i);
    }
    listener(done);
  }, 0);

  return () => {
    clearTimeout(id);
  };
});

// operators.js
import { curry } from "lodash";
import { done } from "./broadcasters";

let createOperator = curry((operator, broadcaster, listener) => {
  return operator((behaviorListener) => {
    return broadcaster((value) => {
      if (value === done) {
        listener(done);
        return;
      }

      behaviorListener(value);
    });
  }, listener);
});

export let map = (transform) =>
  createOperator((broadcaster, listener) => {
    return broadcaster((value) => {
      listener(transform(value));
    });
  });

export let filter = (predicate) =>
  createOperator((broadcaster, listener) => {
    return broadcaster((value) => {
      if (predicate(value)) {
        listener(value);
      }
    });
  });

export let split = (splitter) =>
  curry((broadcaster, listener) => {
    let buffer = [];
    return broadcaster((value) => {
      if (value === done) {
        listener(buffer);
        buffer = [];
        listener(done);
      }
      if (value == splitter) {
        listener(buffer);
        buffer = [];
      } else {
        buffer.push(value);
      }
    });
  });

export let hardCode = (newValue) =>
  createOperator((broadcaster, listener) => {
    let cancel = broadcaster((value) => {
      listener(newValue);
    });

    return () => cancel();
  });

export let add = (initial) => (broadcaster) => (listener) => {
  return broadcaster((value) => {
    listener((initial += value));
  });
};

export let startWhen = (whenBroadcaster) => (mainBroadcaster) => (listener) => {
  let cancelMain;
  let cancelWhen;

  cancelWhen = whenBroadcaster((whenValue) => {
    if (cancelMain) cancelMain();
    cancelMain = mainBroadcaster((value) => {
      if (value === done) {
        if (whenValue === done) {
          listener(done);
        }
        return;
      }
      listener(value);
    });
  });

  return () => {
    cancelMain();
    cancelWhen();
  };
};

export let stopWhen = (whenBroadcaster) => (mainBroadcaster) => (listener) => {
  let cancelMain = mainBroadcaster(listener);

  let cancelWhen = whenBroadcaster((value) => {
    cancelMain();
  });

  return () => {
    cancelMain();
    cancelWhen();
  };
};

export let targetValue = map((event) => event.target.value);

export let mapBroadcaster = (createBroadcaster) => (broadcaster) => (
  listener
) => {
  return broadcaster((value) => {
    let newBroadcaster = createBroadcaster(value);
    newBroadcaster(listener);
  });
};

export let applyOperator = (broadcaster) =>
  mapBroadcaster((operator) => operator(broadcaster));

export let stringConcat = (broadcaster) => (listener) => {
  let result = "";
  return broadcaster((value) => {
    if (value === done) {
      listener(result);
      result = "";
      return;
    }
    result += value;
  });
};

export const sequence = (...broadcasters) => (listener) => {
  let broadcaster = broadcasters.shift();
  let cancel;
  let sequenceListener = (value) => {
    if (value === done && broadcasters.length) {
      broadcaster = broadcasters.shift();
      cancel = broadcaster(sequenceListener);
      return;
    }
    listener(value);
  };
  cancel = broadcaster(sequenceListener);

  return () => {
    cancel();
  };
};

export let doneIf = (condition) => (broadcaster) => (listener) => {
  let cancel;

  cancel = broadcaster((value) => {
    if (condition(value)) {
      listener(done);
      return;
    }

    listener(value);
  });

  return () => {
    cancel();
  };
};