Method pullFromSource

src/share.ts:304–359

* Pull next chunk from source into buffer. * Returns a promise that resolves when the pull completes (or immediately if already pulling).

()

   * Returns a promise that resolves when the pull completes (or immediately if already pulling).
   */
  private pullFromSource(): Promise<void> {
    if (this.sourceExhausted || this.cancelled) {
      return Promise.resolve();
    }

    // If already pulling, wait for that pull to complete
    if (this.pulling) {
      return new Promise<void>((resolve) => {
        this.pullWaiters.push(resolve);
      });
    }

    this.pulling = true;

    return (async () => {
      try {
        // Initialize iterator if needed
        if (!this.sourceIterator) {
          if (isAsyncIterable(this.source)) {
            this.sourceIterator = this.source[Symbol.asyncIterator]();
          } else if (isSyncIterable(this.source)) {
            // Wrap sync iterator
            const syncIterator = (this.source as Iterable<Uint8Array[]>)[Symbol.iterator]();
            this.sourceIterator = {
              async next() {
                return syncIterator.next();
              },
              async return() {
                return syncIterator.return?.() ?? { done: true, value: undefined };
              },
            };
          } else {
            throw new TypeError('Source must be iterable');
          }
        }

        const result = await this.sourceIterator.next();

        if (result.done) {
          this.sourceExhausted = true;
        } else {
          this.buffer.push(result.value);
        }
      } catch (error) {
        this.sourceError = error instanceof Error ? error : new Error(String(error));
        this.sourceExhausted = true;
      } finally {
        this.pulling = false;
        // Wake up waiters so they can check the buffer
        for (const waiter of this.pullWaiters) {
          waiter();
        }
        this.pullWaiters = [];
      }
    })();
  }

  /**