Method #createRawConsumer

lib/internal/streams/iter/share.js:433–566

()

  }

  #createRawConsumer() {
    const state = {
      __proto__: null,
      cursor: this.#bufferStart,
      detached: false,
    };

    this.#consumers.add(state);
    if (this.#consumers.size === 1) {
      this.#cachedMinCursor = state.cursor;
      this.#cachedMinCursorConsumers = 1;
    } else if (state.cursor === this.#cachedMinCursor) {
      this.#cachedMinCursorConsumers++;
    } else {
      this.#recomputeMinCursor();
    }
    const self = this;

    return {
      __proto__: null,
      [SymbolIterator]() {
        return {
          __proto__: null,
          next() {
            if (state.detached) {
              return { __proto__: null, done: true, value: undefined };
            }
            if (self.#sourceError) {
              state.detached = true;
              self.#deleteConsumer(state);
              throw self.#sourceError;
            }
            if (self.#cancelled) {
              state.detached = true;
              self.#deleteConsumer(state);
              return { __proto__: null, done: true, value: undefined };
            }

            const bufferIndex = state.cursor - self.#bufferStart;
            if (bufferIndex < self.#buffer.length) {
              const chunk = self.#buffer.get(bufferIndex);
              const cursor = state.cursor;
              state.cursor++;
              if (cursor === self.#cachedMinCursor &&
                  --self.#cachedMinCursorConsumers === 0) {
                self.#tryTrimBuffer();
              }
              return { __proto__: null, done: false, value: chunk };
            }

            if (self.#sourceExhausted) {
              state.detached = true;
              self.#deleteConsumer(state);
              return { __proto__: null, done: true, value: undefined };
            }

            // Check buffer limit
            if (self.#buffer.length >= self.#options.highWaterMark) {