Synchronous push API to asynchronous generator (event sink)

emitter.ts

export type Emitter<T> = (emit: { (value: T): void }) => Promise<void>;

export function createSink<T>(emitter: Emitter<T>): AsyncIterable<T> {
  const sink: T[] = [];

  let onData = () => { };

  let accept = (value: T) => {
    sink.push(value);
    onData();
  };

  const done = emitter(value => accept(value));

  done.then(() => {
    accept = () => {
      throw new Error(`late value!`); // TODO improve error message
    }
  });

  function asap() {
    if (sink.length) {
      return Promise.resolve();
    }

    const hasData = new Promise<void>(resolve => {
      onData = resolve;
    });

    return Promise.race([hasData, done]);
  }

  return {
    [Symbol.asyncIterator]() {
      return {
        next() {
          return asap().then(() => {
            return sink.length > 0
              ? { done: false, value: sink.shift()! }
              : { done: true, value: undefined };
          });
        }
      }
    }
  }
}