fixup! fs: add FileHandle.prototype.readableWebStream()

doc/api/fs.md

@@ -325,6 +325,8 @@ const file = await open('./some/file/to/read');
 
 for await (const chunk of file.readableWebStream())
   console.log(chunk);
+
+await file.close();
 ```
 
 ```cjs
@@ -337,9 +339,15 @@ const {
 
   for await (const chunk of file.readableWebStream())
     console.log(chunk);
+
+  await file.close();
 })();
 ```
 
+While the `ReadableStream` will read the file to completion, it will not
+close the `FileHandle` automatically. User code must still call the
+`fileHandle.close()` method.
+
 #### `filehandle.readFile(options)`
 <!-- YAML
 added: v10.0.0

lib/internal/fs/promises.js

@@ -102,6 +102,10 @@ const {
   newReadableStreamFromStreamBase,
 } = require('internal/webstreams/adapters');
 
+const {
+  readableStreamCancel,
+} = require('internal/webstreams/readablestream');
+
 const getDirectoryEntriesPromise = promisify(getDirents);
 const validateRmOptionsPromise = promisify(validateRmOptions);
 
@@ -228,7 +232,18 @@ class FileHandle extends EventEmitterMixin(JSTransferable) {
     if (this[kLocked])
       throw new ERR_INVALID_STATE('The FileHandle is locked');
     this[kLocked] = true;
-    return newReadableStreamFromStreamBase(this[kHandle]);
+
+    const readable = newReadableStreamFromStreamBase(
+      this[kHandle],
+      undefined,
+      { ondone: () => this[kUnref]() });
+
+    this[kRef]();
+    this.once('close', () => {
+      readableStreamCancel(readable);
+    });
+
+    return readable;
   }
 
   [kTransfer]() {

lib/internal/webstreams/adapters.js

@@ -859,12 +859,20 @@ function newWritableStreamFromStreamBase(streamBase, strategy) {
  * @param {QueuingStrategy} strategy
  * @returns {ReadableStream}
  */
-function newReadableStreamFromStreamBase(streamBase, strategy) {
+function newReadableStreamFromStreamBase(streamBase, strategy, options = {}) {
   validateObject(streamBase, 'streamBase');
+  validateObject(options, 'options');
+
+  const {
+    ondone = () => {},
+  } = options;
 
   if (typeof streamBase.onread === 'function')
     throw new ERR_INVALID_STATE('StreamBase already has a consumer');
 
+  if (typeof ondone !== 'function')
+    throw new ERR_INVALID_ARG_TYPE('options.ondone', 'Function', ondone);
+
   let controller;
 
   streamBase.onread = (arrayBuffer) => {
@@ -877,6 +885,11 @@ function newReadableStreamFromStreamBase(streamBase, strategy) {
       if (nread === UV_EOF) {
         controller.close();
         streamBase.readStop();
+        try {
+          ondone();
+        } catch (error) {
+          controller.error(error);
+        }
         return;
       }
 
@@ -899,6 +912,12 @@ function newReadableStreamFromStreamBase(streamBase, strategy) {
 
     cancel() {
       const promise = createDeferredPromise();
+      try {
+        ondone();
+      } catch (error) {
+        promise.reject(error);
+        return promise.promise;
+      }
       const req = new ShutdownWrap();
       req.oncomplete = () => promise.resolve();
       const err = streamBase.shutdown(req);

test/parallel/test-filehandle-readablestream.js

@@ -13,6 +13,7 @@ const {
 
 const check = readFileSync(__filename, { encoding: 'utf8' });
 
+// Make sure the ReadableStream works...
 (async () => {
   const dec = new TextDecoder();
   const file = await open(__filename);
@@ -29,6 +30,8 @@ const check = readFileSync(__filename, { encoding: 'utf8' });
   await file.close();
 })().then(common.mustCall());
 
+// Make sure that acquiring a ReadableStream fails if the
+// FileHandle is already closed.
 (async () => {
   const file = await open(__filename);
   await file.close();
@@ -38,6 +41,8 @@ const check = readFileSync(__filename, { encoding: 'utf8' });
   });
 })().then(common.mustCall());
 
+// Make sure that acquiring a ReadableStream fails if the
+// FileHandle is already closing.
 (async () => {
   const file = await open(__filename);
   file.close();
@@ -46,3 +51,37 @@ const check = readFileSync(__filename, { encoding: 'utf8' });
     code: 'ERR_INVALID_STATE',
   });
 })().then(common.mustCall());
+
+// Make sure the ReadableStream is closed when the underlying
+// FileHandle is closed.
+(async () => {
+  const file = await open(__filename);
+  const readable = file.readableWebStream();
+  const reader = readable.getReader();
+  file.close();
+  await reader.closed;
+})().then(common.mustCall());
+
+// Make sure the ReadableStream is closed when the underlying
+// FileHandle is closed.
+(async () => {
+  const file = await open(__filename);
+  const readable = file.readableWebStream();
+  file.close();
+  const reader = readable.getReader();
+  await reader.closed;
+})().then(common.mustCall());
+
+// Make sure that the FileHandle is properly marked "in use"
+// when a ReadableStream has been acquired for it.
+(async () => {
+  const file = await open(__filename);
+  file.readableWebStream();
+  const mc = new MessageChannel();
+  mc.port1.onmessage = common.mustNotCall();
+  assert.throws(() => mc.port2.postMessage(file, [file]), {
+    code: 25  // DataCloneError
+  });
+  mc.port1.close();
+  await file.close();
+})().then(common.mustCall());