stream: add FileHandle support to Read/WriteStream

Support creating a Read/WriteStream from a
FileHandle instead of a raw file descriptor
Add an EventEmitter to FileHandle with a single
'close' event.

Fixes: https://github.com/nodejs/node/issues/35240

PR-URL: https://github.com/nodejs/node/pull/35922
Reviewed-By: Anna Henningsen <anna@addaleax.net>
Reviewed-By: Rich Trott <rtrott@gmail.com>

3 files changed, 108 insertions, 15 deletions

diff --git a/lib/internal/event_target.js b/lib/internal/event_target.js
index c4452b67cf..a0efe8c18e 100644
--- a/lib/internal/event_target.js
+++ b/lib/internal/event_target.js
@@ -4,11 +4,13 @@ const {
   ArrayFrom,
   Boolean,
   Error,
+  FunctionPrototypeCall,
   NumberIsInteger,
   ObjectAssign,
   ObjectDefineProperties,
   ObjectDefineProperty,
   ObjectGetOwnPropertyDescriptor,
+  ObjectGetOwnPropertyDescriptors,
   ReflectApply,
   SafeMap,
   String,
@@ -646,8 +648,23 @@ function defineEventHandler(emitter, name) {
     enumerable: true
   });
 }
+
+const EventEmitterMixin = (Superclass) => {
+  class MixedEventEmitter extends Superclass {
+    constructor(...args) {
+      super(...args);
+      FunctionPrototypeCall(EventEmitter, this);
+    }
+  }
+  const protoProps = ObjectGetOwnPropertyDescriptors(EventEmitter.prototype);
+  delete protoProps.constructor;
+  ObjectDefineProperties(MixedEventEmitter.prototype, protoProps);
+  return MixedEventEmitter;
+};
+
 module.exports = {
   Event,
+  EventEmitterMixin,
   EventTarget,
   NodeEventTarget,
   defineEventHandler,
diff --git a/lib/internal/fs/promises.js b/lib/internal/fs/promises.js
index 93d37a07e7..2f4856e044 100644
--- a/lib/internal/fs/promises.js
+++ b/lib/internal/fs/promises.js
@@ -71,6 +71,7 @@ const {
 } = require('internal/validators');
 const pathModule = require('path');
 const { promisify } = require('internal/util');
+const { EventEmitterMixin } = require('internal/event_target');
 
 const kHandle = Symbol('kHandle');
 const kFd = Symbol('kFd');
@@ -78,6 +79,8 @@ const kRefs = Symbol('kRefs');
 const kClosePromise = Symbol('kClosePromise');
 const kCloseResolve = Symbol('kCloseResolve');
 const kCloseReject = Symbol('kCloseReject');
+const kRef = Symbol('kRef');
+const kUnref = Symbol('kUnref');
 
 const { kUsePromises } = binding;
 const {
@@ -94,7 +97,7 @@ const lazyDOMException = hideStackFrames((message, name) => {
   return new DOMException(message, name);
 });
 
-class FileHandle extends JSTransferable {
+class FileHandle extends EventEmitterMixin(JSTransferable) {
   constructor(filehandle) {
     super();
     this[kHandle] = filehandle;
@@ -197,6 +200,7 @@ class FileHandle extends JSTransferable {
       );
     }
 
+    this.emit('close');
     return this[kClosePromise];
   }
 
@@ -226,6 +230,22 @@ class FileHandle extends JSTransferable {
     this[kHandle] = handle;
     this[kFd] = handle.fd;
   }
+
+  [kRef]() {
+    this[kRefs]++;
+  }
+
+  [kUnref]() {
+    this[kRefs]--;
+    if (this[kRefs] === 0) {
+      this[kFd] = -1;
+      PromisePrototypeThen(
+        this[kHandle].close(),
+        this[kCloseResolve],
+        this[kCloseReject]
+      );
+    }
+  }
 }
 
 async function fsCall(fn, handle, ...args) {
@@ -242,18 +262,10 @@ async function fsCall(fn, handle, ...args) {
   }
 
   try {
-    handle[kRefs]++;
+    handle[kRef]();
     return await fn(handle, ...args);
   } finally {
-    handle[kRefs]--;
-    if (handle[kRefs] === 0) {
-      handle[kFd] = -1;
-      PromisePrototypeThen(
-        handle[kHandle].close(),
-        handle[kCloseResolve],
-        handle[kCloseReject]
-      );
-    }
+    handle[kUnref]();
   }
 }
 
@@ -712,5 +724,7 @@ module.exports = {
     readFile,
   },
 
-  FileHandle
+  FileHandle,
+  kRef,
+  kUnref,
 };
diff --git a/lib/internal/fs/streams.js b/lib/internal/fs/streams.js
index 9dfc7c7325..9542516aa7 100644
--- a/lib/internal/fs/streams.js
+++ b/lib/internal/fs/streams.js
@@ -2,21 +2,25 @@
 
 const {
   Array,
+  FunctionPrototypeBind,
   MathMin,
   ObjectDefineProperty,
   ObjectSetPrototypeOf,
+  PromisePrototypeThen,
   ReflectApply,
   Symbol,
 } = primordials;
 
 const {
   ERR_INVALID_ARG_TYPE,
-  ERR_OUT_OF_RANGE
+  ERR_OUT_OF_RANGE,
+  ERR_METHOD_NOT_IMPLEMENTED,
 } = require('internal/errors').codes;
 const { deprecate } = require('internal/util');
 const { validateInteger } = require('internal/validators');
 const { errorOrDestroy } = require('internal/streams/destroy');
 const fs = require('fs');
+const { kRef, kUnref, FileHandle } = require('internal/fs/promises');
 const { Buffer } = require('buffer');
 const {
   copyObject,
@@ -28,6 +32,7 @@ const kIoDone = Symbol('kIoDone');
 const kIsPerformingIO = Symbol('kIsPerformingIO');
 
 const kFs = Symbol('kFs');
+const kHandle = Symbol('kHandle');
 
 function _construct(callback) {
   const stream = this;
@@ -66,6 +71,35 @@ function _construct(callback) {
   }
 }
 
+// This generates an fs operations structure for a FileHandle
+const FileHandleOperations = (handle) => {
+  return {
+    open: (path, flags, mode, cb) => {
+      throw new ERR_METHOD_NOT_IMPLEMENTED('open()');
+    },
+    close: (fd, cb) => {
+      handle[kUnref]();
+      PromisePrototypeThen(handle.close(),
+                           () => cb(), cb);
+    },
+    read: (fd, buf, offset, length, pos, cb) => {
+      PromisePrototypeThen(handle.read(buf, offset, length, pos),
+                           (r) => cb(null, r.bytesRead, r.buffer),
+                           (err) => cb(err, 0, buf));
+    },
+    write: (fd, buf, offset, length, pos, cb) => {
+      PromisePrototypeThen(handle.write(buf, offset, length, pos),
+                           (r) => cb(null, r.bytesWritten, r.buffer),
+                           (err) => cb(err, 0, buf));
+    },
+    writev: (fd, buffers, pos, cb) => {
+      PromisePrototypeThen(handle.writev(buffers, pos),
+                           (r) => cb(null, r.bytesWritten, r.buffers),
+                           (err) => cb(err, 0, buffers));
+    }
+  };
+};
+
 function close(stream, err, cb) {
   if (!stream.fd) {
     // TODO(ronag)
@@ -80,6 +114,32 @@ function close(stream, err, cb) {
   }
 }
 
+function importFd(stream, options) {
+  stream.fd = null;
+  if (options.fd) {
+    if (typeof options.fd === 'number') {
+      // When fd is a raw descriptor, we must keep our fingers crossed
+      // that the descriptor won't get closed, or worse, replaced with
+      // another one
+      // https://github.com/nodejs/node/issues/35862
+      stream.fd = options.fd;
+    } else if (typeof options.fd === 'object' &&
+      options.fd instanceof FileHandle) {
+      // When fd is a FileHandle we can listen for 'close' events
+      if (options.fs)
+        // FileHandle is not supported with custom fs operations
+        throw new ERR_METHOD_NOT_IMPLEMENTED('FileHandle with fs');
+      stream[kHandle] = options.fd;
+      stream.fd = options.fd.fd;
+      stream[kFs] = FileHandleOperations(stream[kHandle]);
+      stream[kHandle][kRef]();
+      options.fd.on('close', FunctionPrototypeBind(stream.close, stream));
+    } else
+      throw ERR_INVALID_ARG_TYPE('options.fd',
+                                 ['number', 'FileHandle'], options.fd);
+  }
+}
+
 function ReadStream(path, options) {
   if (!(this instanceof ReadStream))
     return new ReadStream(path, options);
@@ -115,10 +175,11 @@ function ReadStream(path, options) {
 
   // Path will be ignored when fd is specified, so it can be falsy
   this.path = toPathIfFileURL(path);
-  this.fd = options.fd === undefined ? null : options.fd;
   this.flags = options.flags === undefined ? 'r' : options.flags;
   this.mode = options.mode === undefined ? 0o666 : options.mode;
 
+  importFd(this, options);
+
   this.start = options.start;
   this.end = options.end;
   this.pos = undefined;
@@ -287,10 +348,11 @@ function WriteStream(path, options) {
 
   // Path will be ignored when fd is specified, so it can be falsy
   this.path = toPathIfFileURL(path);
-  this.fd = options.fd === undefined ? null : options.fd;
   this.flags = options.flags === undefined ? 'w' : options.flags;
   this.mode = options.mode === undefined ? 0o666 : options.mode;
 
+  importFd(this, options);
+
   this.start = options.start;
   this.pos = undefined;
   this.bytesWritten = 0;