fs: implement byob mode for readableWebStream()

Fixes: #45853
PR-URL: #46933
Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>

Author: debadree25

doc/api/fs.md

@@ -444,14 +444,22 @@ 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()`
+#### `filehandle.readableWebStream(options)`
 
 <!-- YAML
 added: v17.0.0
+changes:
+ - version: REPLACEME
+ pr-url: https://github.com/nodejs/node/pull/46933
+ description: Added option to create a 'bytes' stream.
 -->
 
 > Stability: 1 - Experimental
 
+* `options` {Object}
+ * `type` {string|undefined} Whether to open a normal or a `'bytes'` stream.
+ **Default:** `undefined`
+
 * Returns: {ReadableStream}
 
 Returns a `ReadableStream` that may be used to read the files data.

lib/internal/fs/promises.js

@@ -14,6 +14,7 @@ const {
  SafePromisePrototypeFinally,
  Symbol,
  Uint8Array,
+ FunctionPrototypeBind,
 }=primordials;
 
 const{fs: constants}=internalBinding('constants');
@@ -252,7 +253,7 @@ class FileHandle extends EventEmitterMixin(JSTransferable) {
  * } ReadableStream
  * @returns {ReadableStream}
  */
-readableWebStream(){
+readableWebStream(options=kEmptyObject){
 if(this[kFd]===-1)
 thrownewERR_INVALID_STATE('The FileHandle is closed');
 if(this[kClosePromise])
@@ -261,15 +262,64 @@ class FileHandle extends EventEmitterMixin(JSTransferable) {
 thrownewERR_INVALID_STATE('The FileHandle is locked');
 this[kLocked]=true;
 
-constreadable=newReadableStreamFromStreamBase(
-this[kHandle],
-undefined,
-{ondone: ()=>this[kUnref]()});
+if(options.type!==undefined){
+validateString(options.type,'options.type');
+}
 
-this[kRef]();
-this.once('close',()=>{
-readableStreamCancel(readable);
-});
+letreadable;
+
+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');
+
+constreadFn=FunctionPrototypeBind(this.read,this);
+constondone=FunctionPrototypeBind(this[kUnref],this);
+
+readable=newReadableStream({
+type: 'bytes',
+autoAllocateChunkSize: 16384,
+
+asyncpull(controller){
+constview=controller.byobRequest.view;
+const{ bytesRead }=awaitreadFn(view,view.byteOffset,view.byteLength);
+
+if(bytesRead===0){
+ondone();
+controller.close();
+}
+
+controller.byobRequest.respond(bytesRead);
+},
+
+cancel(){
+ondone();
+},
+});
+
+this[kRef]();
+
+this.once('close',()=>{
+readableStreamCancel(readable);
+});
+}
 
 returnreadable;
 }

test/parallel/test-filehandle-readablestream.js

@@ -86,3 +86,87 @@ const check = readFileSync(__filename, { encoding: 'utf8' });
 mc.port1.close();
 awaitfile.close();
 })().then(common.mustCall());
+
+// Make sure 'bytes' stream works
+(async()=>{
+constfile=awaitopen(__filename);
+constdec=newTextDecoder();
+constreadable=file.readableWebStream({type: 'bytes'});
+constreader=readable.getReader({mode: 'byob'});
+
+letdata='';
+letresult;
+do{
+constbuff=newArrayBuffer(100);
+result=awaitreader.read(newDataView(buff));
+if(result.value!==undefined){
+data+=dec.decode(result.value);
+assert.ok(result.value.byteLength<=100);
+}
+}while(!result.done);
+
+assert.strictEqual(check,data);
+
+assert.throws(()=>file.readableWebStream(),{
+code: 'ERR_INVALID_STATE',
+});
+
+awaitfile.close();
+})().then(common.mustCall());
+
+// Make sure that acquiring a ReadableStream 'bytes' stream
+// fails if the FileHandle is already closed.
+(async()=>{
+constfile=awaitopen(__filename);
+awaitfile.close();
+
+assert.throws(()=>file.readableWebStream({type: 'bytes'}),{
+code: 'ERR_INVALID_STATE',
+});
+})().then(common.mustCall());
+
+// Make sure that acquiring a ReadableStream 'bytes' stream
+// fails if the FileHandle is already closing.
+(async()=>{
+constfile=awaitopen(__filename);
+file.close();
+
+assert.throws(()=>file.readableWebStream({type: 'bytes'}),{
+code: 'ERR_INVALID_STATE',
+});
+})().then(common.mustCall());
+
+// Make sure the 'bytes' ReadableStream is closed when the underlying
+// FileHandle is closed.
+(async()=>{
+constfile=awaitopen(__filename);
+constreadable=file.readableWebStream({type: 'bytes'});
+constreader=readable.getReader({mode: 'byob'});
+file.close();
+awaitreader.closed;
+})().then(common.mustCall());
+
+// Make sure the 'bytes' ReadableStream is closed when the underlying
+// FileHandle is closed.
+(async()=>{
+constfile=awaitopen(__filename);
+constreadable=file.readableWebStream({type: 'bytes'});
+file.close();
+constreader=readable.getReader({mode: 'byob'});
+awaitreader.closed;
+})().then(common.mustCall());
+
+// Make sure that the FileHandle is properly marked "in use"
+// when a 'bytes' ReadableStream has been acquired for it.
+(async()=>{
+constfile=awaitopen(__filename);
+file.readableWebStream({type: 'bytes'});
+constmc=newMessageChannel();
+mc.port1.onmessage=common.mustNotCall();
+assert.throws(()=>mc.port2.postMessage(file,[file]),{
+code: 25,
+name: 'DataCloneError',
+});
+mc.port1.close();
+awaitfile.close();
+})().then(common.mustCall());