PDF.js 源码解读: 网络流

class BasePDFStream {
  _fullReader = null;
  _rangeReaders = new Set();
 
  getFullReader() {
    assert(!this._fullReader, "... can only be called once.");
    return (this._fullReader = new this.#PDFStreamReader(this));
  }
 
  getRangeReader(begin, end) {
    if (end <= this._progressiveDataLength) {
      return null; // 全量 reader 已经读过这段，不再发 range 请求
    }
    const reader = new this.#PDFStreamRangeReader(this, begin, end);
    this._rangeReaders.add(reader);
    return reader;
  }
}

class BasePDFStreamReader {
  onProgress = null;
  _contentLength = 0;
  _headersCapability = Promise.withResolvers();
  _isRangeSupported = false;
  _isStreamingSupported = false;
  _loaded = 0;
 
  get headersReady() { return this._headersCapability.promise; }
  get contentLength() { return this._contentLength; }
  get isRangeSupported() { return this._isRangeSupported; }
  get isStreamingSupported() { return this._isStreamingSupported; }
 
  async read() { unreachable("Abstract method `read` called"); }
  cancel(reason) { unreachable("Abstract method `cancel` called"); }
}

function getNetworkStream(url) {
  return isValidFetchUrl(url)
    ? PDFFetchStream
    : typeof PDFJSDev !== "undefined" && PDFJSDev.test("GENERIC") && isNodeJS
      ? PDFNodeStream
      : PDFNetworkStream;
}

class PDFFetchStreamReader extends BasePDFStreamReader {
  _abortController = new AbortController();
  _reader = null;
 
  constructor(stream) {
    super(stream);
    const { disableRange, disableStream, rangeChunkSize, url, withCredentials } =
      stream._source;
 
    this._isStreamingSupported = !disableStream;
    const headers = new Headers(stream.headers); // 永远复制一份
 
    fetchUrl(url, headers, withCredentials, this._abortController)
      .then(response => {
        stream._responseOrigin = getResponseOrigin(response.url);
        ensureResponseStatus(response.status, url); // 只接受 200 / 206
        this._reader = response.body.getReader();
 
        const { contentLength, isRangeSupported } =
          validateRangeRequestCapabilities({
            responseHeaders: response.headers,
            isHttp: true, rangeChunkSize, disableRange,
          });
        this._contentLength = contentLength;
        this._isRangeSupported = isRangeSupported;
        this._filename = extractFilenameFromHeader(response.headers);
 
        // 流式被禁用且支持 range：中止全量请求，之后全部走 range
        if (!this._isStreamingSupported && this._isRangeSupported) {
          this.cancel(new AbortException("Streaming is disabled."));
        }
        this._headersCapability.resolve();
      })
      .catch(this._headersCapability.reject);
  }
}

async read() {
  await this._headersCapability.promise;
  const { value, done } = await this._reader.read();
  if (done) return { value, done };
  this._loaded += value.byteLength;
  this._callOnProgress();
  return { value: getArrayBuffer(value), done: false };
}
 
cancel(reason) {
  this._reader?.cancel(reason);
  this._abortController.abort();
}

class PDFFetchStreamRangeReader extends BasePDFStreamRangeReader {
  _readCapability = Promise.withResolvers();
 
  constructor(stream, begin, end) {
    super(stream, begin, end);
    const headers = new Headers(stream.headers);
    headers.append("Range", `bytes=${begin}-${end - 1}`);
 
    fetchUrl(url, headers, withCredentials, this._abortController)
      .then(response => {
        const responseOrigin = getResponseOrigin(response.url);
        ensureResponseOrigin(responseOrigin, stream._responseOrigin);
        ensureResponseStatus(response.status, url);
        this._reader = response.body.getReader();
        this._readCapability.resolve();
      })
      .catch(this._readCapability.reject);
  }
}

function validateRangeRequestCapabilities({
  responseHeaders, isHttp, rangeChunkSize, disableRange,
}) {
  const rv = { contentLength: 0, isRangeSupported: false };
 
  const length = parseInt(responseHeaders.get("Content-Length"), 10);
  if (!Number.isInteger(length)) return rv;          // 闸门 1
  rv.contentLength = length;
 
  if (length <= 2 * rangeChunkSize) return rv;       // 闸门 2
  if (disableRange || !isHttp) return rv;            // 闸门 3
  if (responseHeaders.get("Accept-Ranges") !== "bytes") return rv; // 闸门 4
 
  const contentEncoding = responseHeaders.get("Content-Encoding") || "identity";
  if (contentEncoding === "identity") {
    rv.isRangeSupported = true;                      // 闸门 5
  }
  return rv;
}

_request(args) {
  const xhr = new XMLHttpRequest();
  const pendingRequest = {
    validateStatus: null,
    onHeadersReceived: args.onHeadersReceived,
    onDone: args.onDone, onError: args.onError, onProgress: args.onProgress,
  };
  this.#pendingRequests.set(xhr, pendingRequest); // WeakMap
 
  xhr.open("GET", this.url);
  xhr.withCredentials = this._source.withCredentials;
  for (const [key, val] of this.headers) xhr.setRequestHeader(key, val);
  if (this.isHttp && "begin" in args && "end" in args) {
    xhr.setRequestHeader("Range", `bytes=${args.begin}-${args.end - 1}`);
    // RFC 2616: "A server MAY ignore the Range header"——可能收到 200 而非 206
    pendingRequest.validateStatus = status =>
      status === PARTIAL_CONTENT_RESPONSE || status === OK_RESPONSE;
  } else {
    pendingRequest.validateStatus = status => status === OK_RESPONSE;
  }
  xhr.responseType = "arraybuffer";
  // ...
}

#onHeadersReceived() {
  // ... 手动把 getAllResponseHeaders() 的字符串解析成 Headers 对象
  const { contentLength, isRangeSupported } = validateRangeRequestCapabilities({ ... });
  this._contentLength = contentLength;
  this._isRangeSupported = isRangeSupported;
 
  if (this._isRangeSupported) {
    // 注释：对"只允许请求一次"的服务器这会出问题——
    // 但那种服务器本来就不该声明支持 range
    stream._abortRequest(fullRequestXhr);
  }
  this._headersCapability.resolve();
}

async read() {
  await this._headersCapability.promise;
  if (this._storedError) throw this._storedError;
  if (this._cachedChunks.length > 0) {
    return { value: this._cachedChunks.shift(), done: false };
  }
  if (this._done) return { value: undefined, done: true };
  const capability = Promise.withResolvers();
  this._requests.push(capability);
  return capability.promise;
}

getRangeReader(begin, end) {
  const reader = super.getRangeReader(begin, end);
  if (reader) {
    reader.onClosed = () => this._rangeReaders.delete(reader);
  }
  return reader;
}

const listener = args => {
  switch (args.type) {
    case "range":
    case "progressiveRead":
      this.#onReceiveData(args.begin, args.chunk);
      break;
    case "progressiveDone":
      this._fullReader?.progressiveDone();
      this._progressiveDone = true;
      break;
  }
};
pdfDataRangeTransport.transportReady(listener);

#onReceiveData(begin, chunk) {
  const buffer = getArrayBuffer(chunk);
  if (begin === undefined) {
    // 渐进数据：给全量 reader（还没建就先排队在 stream 上）
    if (this._fullReader) this._fullReader._enqueue(buffer);
    else this._queuedChunks.push(buffer);
  } else {
    // range 数据：按 begin 找到对应的 range reader
    const rangeReader = this._rangeReaders.keys().find(r => r._begin === begin);
    assert(rangeReader, "...");
    rangeReader._enqueue(buffer);
  }
}

getRangeReader(begin, end) {
  const reader = super.getRangeReader(begin, end);
  if (reader) {
    reader.onDone = () => this._rangeReaders.delete(reader);
    this._source.pdfDataRangeTransport.requestDataRange(begin, end); // 通知子类去拉
  }
  return reader;
}

this._contentLength = length;                          // 来自 transport 构造参数
this._isStreamingSupported = !disableStream;
this._isRangeSupported = !disableRange;

_enqueue(chunk) {
  if (this._done) return;
  if (this._requests.length === 0) {
    this._queuedChunk = chunk;
  } else {
    const capability = this._requests.shift();
    capability.resolve({ value: chunk, done: false });
    this.#endRequests(); // 其余挂起的 read 全部以 done:true 了结
  }
  this._done = true;
  this.onDone?.();
}

messageHandler.on("GetReader", (data, sink) => {
  this.#fullReader = this.#networkStream.getFullReader();
  this.#fullReader.onProgress = evt => this.#onProgress(evt);
 
  sink.onPull = () => {
    this.#fullReader.read().then(({ value, done }) => {
      if (done) { sink.close(); return; }
      sink.enqueue(new Uint8Array(value), 1, [value]); // ArrayBuffer 转移给 worker
    });
  };
  sink.onCancel = reason => this.#fullReader.cancel(reason);
});
 
messageHandler.on("GetRangeReader", (data, sink) => {
  const rangeReader = this.#networkStream.getRangeReader(data.begin, data.end);
  if (!rangeReader) {
    sink.close(); // 数据已在全量流里，直接关流，worker 等全量流送达
    return;
  }
  // onPull / onCancel 与 GetReader 同构
});

messageHandler.on("ReaderHeadersReady", async data => {
  await this.#fullReader.headersReady;
  const { isStreamingSupported, isRangeSupported, contentLength } = this.#fullReader;
  if (isStreamingSupported && isRangeSupported) {
    this.#fullReader.onProgress = null; // range 模式下进度改由别处汇报
  }
  return { isStreamingSupported, isRangeSupported, contentLength };
});