Merge pull request #20594 from Snuffleupagus/Node-ReadableStream

[Node.js] Don't abort the full request for local PDF files smaller than two range requests, and use standard `ReadableStream`s

Author: timvandermeij

gulpfile.mjs

@@ -2366,6 +2366,7 @@ function packageJson() {
     license: DIST_LICENSE,
     optionalDependencies: {
       "@napi-rs/canvas": "^0.1.88",
+      "node-readable-to-web-readable-stream": "^0.4.2",
     },
     browser: {
       canvas: false,

package-lock.json

@@ -39,6 +39,7 @@
     "jstransformer-nunjucks": "^1.2.0",
     "metalsmith": "^2.6.3",
     "metalsmith-html-relative": "^2.0.9",
+    "node-readable-to-web-readable-stream": "^0.4.2",
     "ordered-read-streams": "^2.0.0",
     "pngjs": "^7.0.0",
     "postcss": "^8.5.6",

package.json

@@ -14,7 +14,7 @@
  */
 /* globals process */
 
-import { AbortException, assert } from "../shared/util.js";
+import { AbortException, assert, warn } from "../shared/util.js";
 import { createResponseError } from "./network_utils.js";
 
 if (typeof PDFJSDev !== "undefined" && PDFJSDev.test("MOZCENTRAL")) {
@@ -33,6 +33,33 @@ function parseUrlOrPath(sourceUrl) {
   return new URL(url.pathToFileURL(sourceUrl));
 }
 
+function getReadableStream(readStream) {
+  const { Readable } = process.getBuiltinModule("stream");
+
+  if (typeof Readable.toWeb === "function") {
+    // See https://nodejs.org/api/stream.html#streamreadabletowebstreamreadable-options
+    return Readable.toWeb(readStream);
+  }
+  // Fallback to support Node.js versions older than `24.0.0` and `22.17.0`.
+  const require = process
+    .getBuiltinModule("module")
+    .createRequire(import.meta.url);
+
+  const polyfill = require("node-readable-to-web-readable-stream");
+  return polyfill.makeDefaultReadableStreamFromNodeReadable(readStream);
+}
+
+function getArrayBuffer(val) {
+  if (val instanceof Uint8Array) {
+    return val.buffer;
+  }
+  if (val instanceof ArrayBuffer) {
+    return val;
+  }
+  warn(`getArrayBuffer - unexpected data format: ${val}`);
+  return new Uint8Array(val).buffer;
+}
+
 class PDFNodeStream {
   constructor(source) {
     this.source = source;
@@ -59,11 +86,11 @@ class PDFNodeStream {
     return this._fullRequestReader;
   }
 
-  getRangeReader(start, end) {
+  getRangeReader(begin, end) {
     if (end <= this._progressiveDataLength) {
       return null;
     }
-    const rangeReader = new PDFNodeStreamFsRangeReader(this, start, end);
+    const rangeReader = new PDFNodeStreamFsRangeReader(this, begin, end);
     this._rangeRequestReaders.push(rangeReader);
     return rangeReader;
   }
@@ -78,10 +105,11 @@ class PDFNodeStream {
 }
 
 class PDFNodeStreamFsFullReader {
+  _headersCapability = Promise.withResolvers();
+
+  _reader = null;
+
   constructor(stream) {
-    this._url = stream.url;
-    this._done = false;
-    this._storedError = null;
     this.onProgress = null;
     const source = stream.source;
     this._contentLength = source.length; // optional
@@ -97,27 +125,39 @@ class PDFNodeStreamFsFullReader {
     this._isStreamingSupported = !source.disableStream;
     this._isRangeSupported = !source.disableRange;
 
-    this._readableStream = null;
-    this._readCapability = Promise.withResolvers();
-    this._headersCapability = Promise.withResolvers();
-
+    const url = stream.url;
     const fs = process.getBuiltinModule("fs");
-    fs.promises.lstat(this._url).then(
-      stat => {
+    fs.promises
+      .lstat(url)
+      .then(stat => {
+        const readStream = fs.createReadStream(url);
+        const readableStream = getReadableStream(readStream);
+
+        this._reader = readableStream.getReader();
+
+        const { size } = stat;
+        if (size <= 2 * this._rangeChunkSize) {
+          // The file size is smaller than the size of two chunks, so it doesn't
+          // make any sense to abort the request and retry with a range request.
+          this._isRangeSupported = false;
+        }
         // Setting right content length.
-        this._contentLength = stat.size;
+        this._contentLength = size;
+
+        // We need to stop reading when range is supported and streaming is
+        // disabled.
+        if (!this._isStreamingSupported && this._isRangeSupported) {
+          this.cancel(new AbortException("Streaming is disabled."));
+        }
 
-        this._setReadableStream(fs.createReadStream(this._url));
         this._headersCapability.resolve();
-      },
-      error => {
+      })
+      .catch(error => {
         if (error.code === "ENOENT") {
-          error = createResponseError(/* status = */ 0, this._url.href);
+          error = createResponseError(/* status = */ 0, url.href);
         }
-        this._storedError = error;
         this._headersCapability.reject(error);
-      }
-    );
+      });
   }
 
   get headersReady() {
@@ -141,91 +181,51 @@ class PDFNodeStreamFsFullReader {
   }
 
   async read() {
-    await this._readCapability.promise;
-    if (this._done) {
-      return { value: undefined, done: true };
-    }
-    if (this._storedError) {
-      throw this._storedError;
-    }
-
-    const chunk = this._readableStream.read();
-    if (chunk === null) {
-      this._readCapability = Promise.withResolvers();
-      return this.read();
+    await this._headersCapability.promise;
+    const { value, done } = await this._reader.read();
+    if (done) {
+      return { value, done };
     }
-    this._loaded += chunk.length;
+    this._loaded += value.length;
     this.onProgress?.({
       loaded: this._loaded,
       total: this._contentLength,
     });
 
-    // Ensure that `read()` method returns ArrayBuffer.
-    const buffer = new Uint8Array(chunk).buffer;
-    return { value: buffer, done: false };
+    return { value: getArrayBuffer(value), done: false };
   }
 
   cancel(reason) {
-    // Call `this._error()` method when cancel is called
-    // before _readableStream is set.
-    if (!this._readableStream) {
-      this._error(reason);
-      return;
-    }
-    this._readableStream.destroy(reason);
-  }
-
-  _error(reason) {
-    this._storedError = reason;
-    this._readCapability.resolve();
-  }
-
-  _setReadableStream(readableStream) {
-    this._readableStream = readableStream;
-    readableStream.on("readable", () => {
-      this._readCapability.resolve();
-    });
-
-    readableStream.on("end", () => {
-      // Destroy readable to minimize resource usage.
-      readableStream.destroy();
-      this._done = true;
-      this._readCapability.resolve();
-    });
-
-    readableStream.on("error", reason => {
-      this._error(reason);
-    });
-
-    // We need to stop reading when range is supported and streaming is
-    // disabled.
-    if (!this._isStreamingSupported && this._isRangeSupported) {
-      this._error(new AbortException("streaming is disabled"));
-    }
-
-    // Destroy ReadableStream if already in errored state.
-    if (this._storedError) {
-      this._readableStream.destroy(this._storedError);
-    }
+    this._reader?.cancel(reason);
   }
 }
 
 class PDFNodeStreamFsRangeReader {
-  constructor(stream, start, end) {
-    this._url = stream.url;
-    this._done = false;
-    this._storedError = null;
+  _readCapability = Promise.withResolvers();
+
+  _reader = null;
+
+  constructor(stream, begin, end) {
     this.onProgress = null;
     this._loaded = 0;
-    this._readableStream = null;
-    this._readCapability = Promise.withResolvers();
     const source = stream.source;
     this._isStreamingSupported = !source.disableStream;
 
+    const url = stream.url;
     const fs = process.getBuiltinModule("fs");
-    this._setReadableStream(
-      fs.createReadStream(this._url, { start, end: end - 1 })
-    );
+    try {
+      const readStream = fs.createReadStream(url, {
+        start: begin,
+        end: end - 1,
+      });
+      const readableStream = getReadableStream(readStream);
+
+      this._reader = readableStream.getReader();
+
+      this._readCapability.resolve();
+    } catch (error) {
+      this._readCapability.reject(error);
+    }
   }
 
   get isStreamingSupported() {
@@ -234,62 +234,18 @@ class PDFNodeStreamFsRangeReader {
 
   async read() {
     await this._readCapability.promise;
-    if (this._done) {
-      return { value: undefined, done: true };
-    }
-    if (this._storedError) {
-      throw this._storedError;
-    }
-
-    const chunk = this._readableStream.read();
-    if (chunk === null) {
-      this._readCapability = Promise.withResolvers();
-      return this.read();
+    const { value, done } = await this._reader.read();
+    if (done) {
+      return { value, done };
     }
-    this._loaded += chunk.length;
+    this._loaded += value.length;
     this.onProgress?.({ loaded: this._loaded });
 
-    // Ensure that `read()` method returns ArrayBuffer.
-    const buffer = new Uint8Array(chunk).buffer;
-    return { value: buffer, done: false };
+    return { value: getArrayBuffer(value), done: false };
   }
 
   cancel(reason) {
-    // Call `this._error()` method when cancel is called
-    // before _readableStream is set.
-    if (!this._readableStream) {
-      this._error(reason);
-      return;
-    }
-    this._readableStream.destroy(reason);
-  }
-
-  _error(reason) {
-    this._storedError = reason;
-    this._readCapability.resolve();
-  }
-
-  _setReadableStream(readableStream) {
-    this._readableStream = readableStream;
-    readableStream.on("readable", () => {
-      this._readCapability.resolve();
-    });
-
-    readableStream.on("end", () => {
-      // Destroy readableStream to minimize resource usage.
-      readableStream.destroy();
-      this._done = true;
-      this._readCapability.resolve();
-    });
-
-    readableStream.on("error", reason => {
-      this._error(reason);
-    });
-
-    // Destroy readableStream if already in errored state.
-    if (this._storedError) {
-      this._readableStream.destroy(this._storedError);
-    }
+    this._reader?.cancel(reason);
   }
 }
 

src/display/node_stream.js

@@ -117,4 +117,41 @@ describe("node_stream", function () {
     expect(isRangeSupported).toEqual(true);
     expect(fullReaderCancelled).toEqual(true);
   });
+
+  it("read filesystem pdf files (smaller than two range requests)", async function () {
+    const smallPdf = new URL("./test/pdfs/empty.pdf", cwdURL).href;
+    const smallLength = 4920;
+
+    const stream = new PDFNodeStream({
+      url: smallPdf,
+      rangeChunkSize: 65536,
+      disableStream: true,
+      disableRange: false,
+    });
+
+    const fullReader = stream.getFullReader();
+
+    let isStreamingSupported, isRangeSupported;
+    const promise = fullReader.headersReady.then(() => {
+      isStreamingSupported = fullReader.isStreamingSupported;
+      isRangeSupported = fullReader.isRangeSupported;
+    });
+
+    let len = 0;
+    const read = function () {
+      return fullReader.read().then(function (result) {
+        if (result.done) {
+          return undefined;
+        }
+        len += result.value.byteLength;
+        return read();
+      });
+    };
+
+    await Promise.all([read(), promise]);
+
+    expect(isStreamingSupported).toEqual(false);
+    expect(isRangeSupported).toEqual(false);
+    expect(len).toEqual(smallLength);
+  });
 });