An experiment towards async push/pull streams

async-stream.ts

type AsyncCell<T> = { value: T; next: Promise<AsyncCell<T>> };
type AsyncStreamOutput<T> = () => AsyncGenerator<unknown, T, unknown>;
type AsyncStreamInput<T> = (value: T) => unknown;

const mkPromiseAndInput = <T>(): [Promise<T>, AsyncStreamInput<T>] => {
  // streamInput is assigned before return because of Promise implementation
  let streamInput!: AsyncStreamInput<T>;
  const promise = new Promise<T>((resolve) => {
    streamInput = resolve as AsyncStreamInput<T>;
  });

  return [promise, streamInput];
};

const mkAsyncStream = <T>(): [AsyncStreamOutput<T>, AsyncStreamInput<T>] => {
  let [currentCellPromise, resolveCurrentCell] = mkPromiseAndInput<
    AsyncCell<T>
  >();

  const streamInput: AsyncStreamInput<T> = (value: T) => {
    const resolvePreviousCell = resolveCurrentCell;
    [currentCellPromise, resolveCurrentCell] = mkPromiseAndInput<
      AsyncCell<T>
    >();
    resolvePreviousCell({ value, next: currentCellPromise });
  };

  const streamOutput = async function* () {
    let currentCell = await currentCellPromise;
    while (true) {
      yield currentCell.value;
      currentCell = await currentCell.next;
    }
  };

  return [streamOutput, streamInput];
};

Constraints:

1. Have an input where some events can be pushed into at any time in the future.
2. Support streaming such events using and async generator.
3. Support multiple async generator streams at a time. Support the amount of streams to be changing.
4. Don't accumulate events in storage if no generator is waiting for them.

Wrote a bit more text on it here:
https://www.runjak.codes/posts/2022-08-06-async-stream