Class BroadcastImpl

lib/internal/streams/iter/broadcast.js:85–474

// =============================================================================

class BroadcastImpl {
  #buffer = new RingBuffer();
  #bufferStart = 0;
  #consumers = new SafeSet();
  #waiters = [];  // Consumers with pending resolve (subset of #consumers)
  #ended = false;
  #error = null;
  #cancelled = false;
  #options;
  #writer = null;
  #cachedMinCursor = 0;
  #cachedMinCursorConsumers = 0;

  constructor(options) {
    this.#options = options;
    this[kOnBufferDrained] = null;
  }

  setWriter(writer) {
    this.#writer = writer;
  }

  get backpressurePolicy() {
    return this.#options.backpressure;
  }

  get highWaterMark() {
    return this.#options.highWaterMark;
  }

  get consumerCount() {
    return this.#consumers.size;
  }

  get bufferSize() {
    return this.#buffer.length;
  }

  push(...args) {
    const { transforms, options } = parsePullArgs(args);
    const rawConsumer = this.#createRawConsumer();

    // When transforms are present, delegate to pull() which creates its
    // own internal AbortController that follows the external signal.
    // When no transforms, return rawConsumer directly (controller elided
    // per PULL-02 optimization -- no transforms means no signal recipient).
    if (transforms.length > 0) {
      const pullArgs = [...transforms];
      if (options?.signal) {
        ArrayPrototypePush(pullArgs,
                           { __proto__: null, signal: options.signal });
      }
      return pullWithTransforms(rawConsumer, ...pullArgs);
    }
    return rawConsumer;
  }

  #createRawConsumer() {