fs: add FileHandle.prototype.readableWebStream()

Adds an experimental `readableWebStream()` method to `FileHandle` that
returns a web `ReadableStream`

Signed-off-by: James M Snell <[email protected]>

doc/api/fs.md

@@ -302,6 +302,44 @@ Reads data from the file and stores that in the given buffer.
 If the file is not modified concurrently, the end-of-file is reached when the
 number of bytes read is zero.
 
+#### `filehandle.readableWebStream()`
+<!-- YAML
+added: REPLACEME
+-->
+
+> Stability: 1 - Experimental
+
+* Returns: {ReadableStream}
+
+Returns a `ReadableStream` that may be used to read the files data.
+
+An error will be thrown if this method is called more than once or is called
+after the `FileHandle` is closed or closing.
+
+```mjs
+import {
+  open,
+} from 'node:fs/promises';
+
+const file = await open('./some/file/to/read');
+
+for await (const chunk of file.readableWebStream())
+  console.log(chunk);
+```
+
+```cjs
+const {
+  open,
+} = require('fs/promises');
+
+(async () => {
+  const file = await open('./some/file/to/read');
+
+  for await (const chunk of file.readableWebStream())
+    console.log(chunk);
+})();
+```
+
 #### `filehandle.readFile(options)`
 <!-- YAML
 added: v10.0.0

lib/internal/fs/promises.js

@@ -30,6 +30,7 @@ const {
   codes: {
     ERR_FS_FILE_TOO_LARGE,
     ERR_INVALID_ARG_VALUE,
+    ERR_INVALID_STATE,
     ERR_METHOD_NOT_IMPLEMENTED,
   },
   AbortError,
@@ -90,12 +91,17 @@ const kCloseResolve = Symbol('kCloseResolve');
 const kCloseReject = Symbol('kCloseReject');
 const kRef = Symbol('kRef');
 const kUnref = Symbol('kUnref');
+const kLocked = Symbol('kLocked');
 
 const { kUsePromises } = binding;
 const {
   JSTransferable, kDeserialize, kTransfer, kTransferList
 } = require('internal/worker/js_transferable');
 
+const {
+  newReadableStreamFromStreamBase,
+} = require('internal/webstreams/adapters');
+
 const getDirectoryEntriesPromise = promisify(getDirents);
 const validateRmOptionsPromise = promisify(validateRmOptions);
 
@@ -209,6 +215,22 @@ class FileHandle extends EventEmitterMixin(JSTransferable) {
     return this[kClosePromise];
   }
 
+  /**
+   * @typedef {import('../webstreams/readablestream').ReadableStream
+   * } ReadableStream
+   * @returns {ReadableStream}
+   */
+  readableWebStream() {
+    if (this[kFd] === -1)
+      throw new ERR_INVALID_STATE('The FileHandle is closed');
+    if (this[kClosePromise])
+      throw new ERR_INVALID_STATE('The FileHandle is closing');
+    if (this[kLocked])
+      throw new ERR_INVALID_STATE('The FileHandle is locked');
+    this[kLocked] = true;
+    return newReadableStreamFromStreamBase(this[kHandle]);
+  }
+
   [kTransfer]() {
     if (this[kClosePromise] || this[kRefs] > 1) {
       throw lazyDOMException('Cannot transfer FileHandle while in use',

test/parallel/test-filehandle-readablestream.js

@@ -0,0 +1,48 @@
+'use strict';
+
+const common = require('../common');
+const assert = require('assert');
+
+const {
+  readFileSync,
+} = require('fs');
+
+const {
+  open,
+} = require('fs/promises');
+
+const check = readFileSync(__filename, { encoding: 'utf8' });
+
+(async () => {
+  const dec = new TextDecoder();
+  const file = await open(__filename);
+  let data = '';
+  for await (const chunk of file.readableWebStream())
+    data += dec.decode(chunk);
+
+  assert.strictEqual(check, data);
+
+  assert.throws(() => file.readableWebStream(), {
+    code: 'ERR_INVALID_STATE',
+  });
+
+  await file.close();
+})().then(common.mustCall());
+
+(async () => {
+  const file = await open(__filename);
+  await file.close();
+
+  assert.throws(() => file.readableWebStream(), {
+    code: 'ERR_INVALID_STATE',
+  });
+})().then(common.mustCall());
+
+(async () => {
+  const file = await open(__filename);
+  file.close();
+
+  assert.throws(() => file.readableWebStream(), {
+    code: 'ERR_INVALID_STATE',
+  });
+})().then(common.mustCall());