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>

4 files changed, 159 insertions, 24 deletions

diff --git a/test/parallel/test-fs-promises-file-handle-read-worker.js b/test/parallel/test-fs-promises-file-handle-read-worker.js
new file mode 100644
index 0000000000..ccf3338d14
--- /dev/null
+++ b/test/parallel/test-fs-promises-file-handle-read-worker.js
@@ -0,0 +1,50 @@
+'use strict';
+const common = require('../common');
+const fs = require('fs');
+const assert = require('assert');
+const path = require('path');
+const tmpdir = require('../common/tmpdir');
+const file = path.join(tmpdir.path, 'read_stream_filehandle_worker.txt');
+const input = 'hello world';
+const { Worker, isMainThread, workerData } = require('worker_threads');
+
+if (isMainThread || !workerData) {
+  tmpdir.refresh();
+  fs.writeFileSync(file, input);
+
+  fs.promises.open(file, 'r').then((handle) => {
+    handle.on('close', common.mustNotCall());
+    new Worker(__filename, {
+      workerData: { handle },
+      transferList: [handle]
+    });
+  });
+  fs.promises.open(file, 'r').then((handle) => {
+    fs.createReadStream(null, { fd: handle });
+    assert.throws(() => {
+      new Worker(__filename, {
+        workerData: { handle },
+        transferList: [handle]
+      });
+    }, {
+      code: 25,
+    });
+  });
+} else {
+  let output = '';
+
+  const handle = workerData.handle;
+  handle.on('close', common.mustCall());
+  const stream = fs.createReadStream(null, { fd: handle });
+
+  stream.on('data', common.mustCallAtLeast((data) => {
+    output += data;
+  }));
+
+  stream.on('end', common.mustCall(() => {
+    handle.close();
+    assert.strictEqual(output, input);
+  }));
+
+  stream.on('close', common.mustCall());
+}
diff --git a/test/parallel/test-fs-promises-file-handle-read.js b/test/parallel/test-fs-promises-file-handle-read.js
index 44c35840f2..68ad5c50ac 100644
--- a/test/parallel/test-fs-promises-file-handle-read.js
+++ b/test/parallel/test-fs-promises-file-handle-read.js
@@ -19,33 +19,32 @@ async function read(fileHandle, buffer, offset, length, position) {
     fileHandle.read(buffer, offset, length, position);
 }
 
-async function validateRead() {
-  const filePath = path.resolve(tmpDir, 'tmp-read-file.txt');
-  const fileHandle = await open(filePath, 'w+');
-  const buffer = Buffer.from('Hello world', 'utf8');
+async function validateRead(data, file) {
+  const filePath = path.resolve(tmpDir, file);
+  const buffer = Buffer.from(data, 'utf8');
 
   const fd = fs.openSync(filePath, 'w+');
-  fs.writeSync(fd, buffer, 0, buffer.length);
-  fs.closeSync(fd);
-  const readAsyncHandle = await read(fileHandle, Buffer.alloc(11), 0, 11, 0);
-  assert.deepStrictEqual(buffer.length, readAsyncHandle.bytesRead);
-  assert.deepStrictEqual(buffer, readAsyncHandle.buffer);
-
-  await fileHandle.close();
-}
-
-async function validateEmptyRead() {
-  const filePath = path.resolve(tmpDir, 'tmp-read-empty-file.txt');
   const fileHandle = await open(filePath, 'w+');
-  const buffer = Buffer.from('', 'utf8');
+  const streamFileHandle = await open(filePath, 'w+');
 
-  const fd = fs.openSync(filePath, 'w+');
   fs.writeSync(fd, buffer, 0, buffer.length);
   fs.closeSync(fd);
-  const readAsyncHandle = await read(fileHandle, Buffer.alloc(11), 0, 11, 0);
-  assert.deepStrictEqual(buffer.length, readAsyncHandle.bytesRead);
 
+  fileHandle.on('close', common.mustCall());
+  const readAsyncHandle = await read(fileHandle, Buffer.alloc(11), 0, 11, 0);
+  assert.deepStrictEqual(data.length, readAsyncHandle.bytesRead);
+  if (data.length)
+    assert.deepStrictEqual(buffer, readAsyncHandle.buffer);
   await fileHandle.close();
+
+  const stream = fs.createReadStream(null, { fd: streamFileHandle });
+  let streamData = Buffer.alloc(0);
+  for await (const chunk of stream)
+    streamData = Buffer.from(chunk);
+  assert.deepStrictEqual(buffer, streamData);
+  if (data.length)
+    assert.deepStrictEqual(streamData, readAsyncHandle.buffer);
+  await streamFileHandle.close();
 }
 
 async function validateLargeRead() {
@@ -67,9 +66,9 @@ let useConf = false;
     tmpdir.refresh();
     useConf = value;
 
-    await validateRead()
-          .then(validateEmptyRead)
-          .then(validateLargeRead)
-          .then(common.mustCall());
+    await validateRead('Hello world', 'tmp-read-file.txt')
+      .then(validateRead('', 'tmp-read-empty-file.txt'))
+      .then(validateLargeRead)
+      .then(common.mustCall());
   }
-});
+})().then(common.mustCall());
diff --git a/test/parallel/test-fs-read-stream-file-handle.js b/test/parallel/test-fs-read-stream-file-handle.js
new file mode 100644
index 0000000000..063075aa25
--- /dev/null
+++ b/test/parallel/test-fs-read-stream-file-handle.js
@@ -0,0 +1,65 @@
+'use strict';
+const common = require('../common');
+const fs = require('fs');
+const assert = require('assert');
+const path = require('path');
+const tmpdir = require('../common/tmpdir');
+const file = path.join(tmpdir.path, 'read_stream_filehandle_test.txt');
+const input = 'hello world';
+
+let output = '';
+tmpdir.refresh();
+fs.writeFileSync(file, input);
+
+fs.promises.open(file, 'r').then(common.mustCall((handle) => {
+  handle.on('close', common.mustCall());
+  const stream = fs.createReadStream(null, { fd: handle });
+
+  stream.on('data', common.mustCallAtLeast((data) => {
+    output += data;
+  }));
+
+  stream.on('end', common.mustCall(() => {
+    assert.strictEqual(output, input);
+  }));
+
+  stream.on('close', common.mustCall());
+}));
+
+fs.promises.open(file, 'r').then(common.mustCall((handle) => {
+  handle.on('close', common.mustCall());
+  const stream = fs.createReadStream(null, { fd: handle });
+  stream.on('data', common.mustNotCall());
+  stream.on('close', common.mustCall());
+
+  handle.close();
+}));
+
+fs.promises.open(file, 'r').then(common.mustCall((handle) => {
+  handle.on('close', common.mustCall());
+  const stream = fs.createReadStream(null, { fd: handle });
+  stream.on('close', common.mustCall());
+
+  stream.on('data', common.mustCall(() => {
+    handle.close();
+  }));
+}));
+
+fs.promises.open(file, 'r').then(common.mustCall((handle) => {
+  handle.on('close', common.mustCall());
+  const stream = fs.createReadStream(null, { fd: handle });
+  stream.on('close', common.mustCall());
+
+  stream.close();
+}));
+
+fs.promises.open(file, 'r').then(common.mustCall((handle) => {
+  assert.throws(() => {
+    fs.createReadStream(null, { fd: handle, fs });
+  }, {
+    code: 'ERR_METHOD_NOT_IMPLEMENTED',
+    name: 'Error',
+    message: 'The FileHandle with fs method is not implemented'
+  });
+  handle.close();
+}));
diff --git a/test/parallel/test-fs-write-stream-file-handle.js b/test/parallel/test-fs-write-stream-file-handle.js
new file mode 100644
index 0000000000..37d9a81d7c
--- /dev/null
+++ b/test/parallel/test-fs-write-stream-file-handle.js
@@ -0,0 +1,21 @@
+'use strict';
+const common = require('../common');
+const fs = require('fs');
+const path = require('path');
+const assert = require('assert');
+const tmpdir = require('../common/tmpdir');
+const file = path.join(tmpdir.path, 'write_stream_filehandle_test.txt');
+const input = 'hello world';
+
+tmpdir.refresh();
+
+fs.promises.open(file, 'w+').then(common.mustCall((handle) => {
+  handle.on('close', common.mustCall());
+  const stream = fs.createWriteStream(null, { fd: handle });
+
+  stream.end(input);
+  stream.on('close', common.mustCall(() => {
+    const output = fs.readFileSync(file, 'utf-8');
+    assert.strictEqual(output, input);
+  }));
+}));