事件流单槽交接与背压

策略的具体实现如下。

export async function* streamEvents<Event>(
  operation: (emit: (event: Event) => Promise<void>) => Promise<void>,
  { signal }: import('../structs/common.js').ExecutionContext,
): AsyncIterable<Event> {
  let slot: { event: Event; release(): void; cancel(): void } | undefined;
  let wake: (() => void) | undefined;
  let closed = false,
    done = false;
  let sending = false;
  let failure: { error: unknown } | undefined;
  let rejectSend: ((error: unknown) => void) | undefined;
  const notify = () => {
    wake?.();
    wake = undefined;
  };
  const abort = () => {
    closed = true;
    rejectSend?.(signal.reason ?? new Error('Stream closed'));
    notify();
  };
  signal.addEventListener('abort', abort, { once: true });
  if (signal.aborted) abort();
  const producer = Promise.resolve()
    .then(() => {
      signal.throwIfAborted();
      // Before the producer starts, only abort can close this rendezvous; the
      // signal check above already rejects it with the original reason.
      return operation((event) => {
        if (closed || signal.aborted)
          return Promise.reject(signal.reason ?? new Error('Stream closed'));
        if (sending) return Promise.reject(new Error('Event producers must await delivery'));
        sending = true;
        return new Promise<void>((resolve, reject) => {
          rejectSend = reject;
          slot = {
            event,
            release: () => {
              sending = false;
              resolve();
            },
            cancel: () => {
              sending = false;
              reject(new Error('Stream closed before event consumption'));
            },
          };
          notify();
        });
      });
    })
    .catch((error) => {
      failure = { error };
    })
    .finally(() => {
      done = true;
      notify();
    });
  try {
    while (true) {
      signal.throwIfAborted();
      if (slot) {
        const current = slot;
        slot = undefined;
        let consumed = false;
        try {
          yield current.event;
          consumed = true;
        } finally {
          if (consumed) current.release();
          else current.cancel();
        }
      } else if (done) {
        if (failure) throw failure.error;
        return;
      } else
        await new Promise<void>((resolve) => {
          wake = resolve;
        });
    }
  } finally {
    closed = true;
    rejectSend?.(new Error('Stream closed'));
    signal.removeEventListener('abort', abort);
    void producer;
  }
}