fs: implement byob mode for readableWebStream()

Fixes: nodejs#45853

lib/internal/fs/promises.js

@@ -14,6 +14,7 @@ const {
   SafePromisePrototypeFinally,
   Symbol,
   Uint8Array,
+  FunctionPrototypeBind,
 } = primordials;
 
 const { fs: constants } = internalBinding('constants');
@@ -249,29 +250,68 @@ class FileHandle extends EventEmitterMixin(JSTransferable) {
    * } ReadableStream
    * @returns {ReadableStream}
    */
-  readableWebStream() {
+  readableWebStream(options = kEmptyObject) {
     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;
-    const {
-      newReadableStreamFromStreamBase,
-    } = require('internal/webstreams/adapters');
-    const readable = newReadableStreamFromStreamBase(
-      this[kHandle],
-      undefined,
-      { ondone: () => this[kUnref]() });
-
-    const {
-      readableStreamCancel,
-    } = require('internal/webstreams/readablestream');
-    this[kRef]();
-    this.once('close', () => {
-      readableStreamCancel(readable);
-    });
+
+    if (options.type !== undefined) {
+      validateString(options.type, 'options.type');
+    }
+
+    let readable;
+
+    if (options.type !== 'bytes') {
+      const {
+        newReadableStreamFromStreamBase,
+      } = require('internal/webstreams/adapters');
+      readable = newReadableStreamFromStreamBase(
+        this[kHandle],
+        undefined,
+        { ondone: () => this[kUnref]() });
+
+      const {
+        readableStreamCancel,
+      } = require('internal/webstreams/readablestream');
+      this[kRef]();
+      this.once('close', () => {
+        readableStreamCancel(readable);
+      });
+    } else {
+      const {
+        readableStreamCancel,
+        ReadableStream,
+      } = require('internal/webstreams/readablestream');
+      this[kRef]();
+      this.once('close', () => {
+        readableStreamCancel(readable);
+      });
+      const readFn = FunctionPrototypeBind(this.read, this);
+      readable = new ReadableStream({
+        type: 'bytes',
+        autoAllocateChunkSize: 16384,
+
+        async pull(controller) {
+          const view = controller.byobRequest.view;
+          const { bytesRead } = await readFn(view, view.byteOffset, view.byteLength);
+
+          if (bytesRead === 0) {
+            try {
+              controller.close();
+              this[kUnref]();
+            } catch (error) {
+              controller.error(error);
+            }
+          }
+
+          controller.byobRequest.respond(bytesRead);
+        },
+      });
+    }
 
     return readable;
   }