Function onData

src/cmap/wire_protocol/on_data.ts:23–139

(
  emitter: EventEmitter,
  { timeoutContext, signal }: { timeoutContext?: TimeoutContext } & Abortable
)

 * It will reject upon an error event.
 */
export function onData(
  emitter: EventEmitter,
  { timeoutContext, signal }: { timeoutContext?: TimeoutContext } & Abortable
) {
  signal?.throwIfAborted();

  // Setup pending events and pending promise lists
  /**
   * When the caller has not yet called .next(), we store the
   * value from the event in this list. Next time they call .next()
   * we pull the first value out of this list and resolve a promise with it.
   */
  const unconsumedEvents = new List<Uint8Array>();
  /**
   * When there has not yet been an event, a new promise will be created
   * and implicitly stored in this list. When an event occurs we take the first
   * promise in this list and resolve it.
   */
  const unconsumedPromises = new List<PendingPromises>();

  /**
   * Stored an error created by an error event.
   * This error will turn into a rejection for the subsequent .next() call
   */
  let error: Error | null = null;

  /** Set to true only after event listeners have been removed. */
  let finished = false;

  const iterator: AsyncGenerator<Uint8Array> & AsyncDisposable = {
    next() {
      // First, we consume all unread events
      const value = unconsumedEvents.shift();
      if (value != null) {
        return Promise.resolve({ value, done: false });
      }

      // Then we error, if an error happened
      // This happens one time if at all, because after 'error'
      // we stop listening
      if (error != null) {
        const p = Promise.reject(error);
        // Only the first element errors
        error = null;
        return p;
      }

      // If the iterator is finished, resolve to done
      if (finished) return closeHandler();

      // Wait until an event happens
      const { promise, resolve, reject } = promiseWithResolvers<IteratorResult<Uint8Array>>();
      unconsumedPromises.push({ resolve, reject });
      return promise;
    },

    return() {
      return closeHandler();