fs: add support for async iterators to fsPromises.writeFile

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

PR-URL: https://github.com/nodejs/node/pull/37490
Reviewed-By: Antoine du Hamel <duhamelantoine1995@gmail.com>
Reviewed-By: James M Snell <jasnell@gmail.com>

doc/api/fs.md

@@ -1252,6 +1252,9 @@ All the [caveats][] for `fs.watch()` also apply to `fsPromises.watch()`.
 <!-- YAML
 added: v10.0.0
 changes:
+  - version: REPLACEME
+    pr-url: https://github.com/nodejs/node/pull/37490
+    description: The `data` argument supports `AsyncIterable`, `Iterable` & `Stream`.
   - version: v15.2.0
     pr-url: https://github.com/nodejs/node/pull/35993
     description: The options argument may include an AbortSignal to abort an
@@ -1267,7 +1270,8 @@ changes:
 -->
 
 * `file` {string|Buffer|URL|FileHandle} filename or `FileHandle`
-* `data` {string|Buffer|Uint8Array|Object}
+* `data` {string|Buffer|Uint8Array|Object|AsyncIterable|Iterable
+  |Stream}
 * `options` {Object|string}
   * `encoding` {string|null} **Default:** `'utf8'`
   * `mode` {integer} **Default:** `0o666`

lib/internal/fs/promises.js

@@ -80,6 +80,7 @@ const pathModule = require('path');
 const { promisify } = require('internal/util');
 const { EventEmitterMixin } = require('internal/event_target');
 const { watch } = require('internal/fs/watchers');
+const { isIterable } = require('internal/streams/utils');
 
 const kHandle = Symbol('kHandle');
 const kFd = Symbol('kFd');
@@ -275,8 +276,18 @@ function checkAborted(signal) {
     throw new AbortError();
 }
 
-async function writeFileHandle(filehandle, data, signal) {
-  // `data` could be any kind of typed array.
+async function writeFileHandle(filehandle, data, signal, encoding) {
+  checkAborted(signal);
+  if (isCustomIterable(data)) {
+    for await (const buf of data) {
+      checkAborted(signal);
+      await write(
+        filehandle, buf, undefined,
+        isArrayBufferView(buf) ? buf.length : encoding);
+      checkAborted(signal);
+    }
+    return;
+  }
   data = new Uint8Array(data.buffer, data.byteOffset, data.byteLength);
   let remaining = data.length;
   if (remaining === 0) return;
@@ -439,7 +450,7 @@ async function readv(handle, buffers, position) {
 }
 
 async function write(handle, buffer, offset, length, position) {
-  if (buffer.length === 0)
+  if (buffer?.length === 0)
     return { bytesWritten: 0, buffer };
 
   if (isArrayBufferView(buffer)) {
@@ -684,20 +695,24 @@ async function writeFile(path, data, options) {
   options = getOptions(options, { encoding: 'utf8', mode: 0o666, flag: 'w' });
   const flag = options.flag || 'w';
 
-  if (!isArrayBufferView(data)) {
+  if (!isArrayBufferView(data) && !isCustomIterable(data)) {
     validateStringAfterArrayBufferView(data, 'data');
     data = Buffer.from(data, options.encoding || 'utf8');
   }
 
   validateAbortSignal(options.signal);
   if (path instanceof FileHandle)
-    return writeFileHandle(path, data, options.signal);
+    return writeFileHandle(path, data, options.signal, options.encoding);
 
   checkAborted(options.signal);
 
   const fd = await open(path, flag, options.mode);
-  const { signal } = options;
-  return PromisePrototypeFinally(writeFileHandle(fd, data, signal), fd.close);
+  return PromisePrototypeFinally(
+    writeFileHandle(fd, data, options.signal, options.encoding), fd.close);
+}
+
+function isCustomIterable(obj) {
+  return isIterable(obj) && !isArrayBufferView(obj) && typeof obj !== 'string';
 }
 
 async function appendFile(path, data, options) {

test/parallel/test-fs-append-file.js

@@ -121,7 +121,7 @@ const throwNextTick = (e) => { process.nextTick(() => { throw e; }); };
 }
 
 // Test that appendFile does not accept invalid data type (callback API).
-[false, 5, {}, [], null, undefined].forEach(async (data) => {
+[false, 5, {}, null, undefined].forEach(async (data) => {
   const errObj = {
     code: 'ERR_INVALID_ARG_TYPE',
     message: /"data"|"buffer"/

test/parallel/test-fs-promises-writefile.js

@@ -7,6 +7,7 @@ const path = require('path');
 const tmpdir = require('../common/tmpdir');
 const assert = require('assert');
 const tmpDir = tmpdir.path;
+const { Readable } = require('stream');
 
 tmpdir.refresh();
 
@@ -14,6 +15,39 @@ const dest = path.resolve(tmpDir, 'tmp.txt');
 const otherDest = path.resolve(tmpDir, 'tmp-2.txt');
 const buffer = Buffer.from('abc'.repeat(1000));
 const buffer2 = Buffer.from('xyz'.repeat(1000));
+const stream = Readable.from(['a', 'b', 'c']);
+const stream2 = Readable.from(['ümlaut', ' ', 'sechzig']);
+const iterable = {
+  expected: 'abc',
+  *[Symbol.iterator]() {
+    yield 'a';
+    yield 'b';
+    yield 'c';
+  }
+};
+function iterableWith(value) {
+  return {
+    *[Symbol.iterator]() {
+      yield value;
+    }
+  };
+}
+const bufferIterable = {
+  expected: 'abc',
+  *[Symbol.iterator]() {
+    yield Buffer.from('a');
+    yield Buffer.from('b');
+    yield Buffer.from('c');
+  }
+};
+const asyncIterable = {
+  expected: 'abc',
+  async* [Symbol.asyncIterator]() {
+    yield 'a';
+    yield 'b';
+    yield 'c';
+  }
+};
 
 async function doWrite() {
   await fsPromises.writeFile(dest, buffer);
@@ -21,6 +55,67 @@ async function doWrite() {
   assert.deepStrictEqual(data, buffer);
 }
 
+async function doWriteStream() {
+  await fsPromises.writeFile(dest, stream);
+  const expected = 'abc';
+  const data = fs.readFileSync(dest, 'utf-8');
+  assert.deepStrictEqual(data, expected);
+}
+
+async function doWriteStreamWithCancel() {
+  const controller = new AbortController();
+  const { signal } = controller;
+  process.nextTick(() => controller.abort());
+  assert.rejects(fsPromises.writeFile(otherDest, stream, { signal }), {
+    name: 'AbortError'
+  });
+}
+
+async function doWriteIterable() {
+  await fsPromises.writeFile(dest, iterable);
+  const data = fs.readFileSync(dest, 'utf-8');
+  assert.deepStrictEqual(data, iterable.expected);
+}
+
+async function doWriteInvalidIterable() {
+  await Promise.all(
+    [42, 42n, {}, Symbol('42'), true, undefined, null, NaN].map((value) =>
+      assert.rejects(fsPromises.writeFile(dest, iterableWith(value)), {
+        code: 'ERR_INVALID_ARG_TYPE',
+      })
+    )
+  );
+}
+
+async function doWriteIterableWithEncoding() {
+  await fsPromises.writeFile(dest, stream2, 'latin1');
+  const expected = 'ümlaut sechzig';
+  const data = fs.readFileSync(dest, 'latin1');
+  assert.deepStrictEqual(data, expected);
+}
+
+async function doWriteBufferIterable() {
+  await fsPromises.writeFile(dest, bufferIterable);
+  const data = fs.readFileSync(dest, 'utf-8');
+  assert.deepStrictEqual(data, bufferIterable.expected);
+}
+
+async function doWriteAsyncIterable() {
+  await fsPromises.writeFile(dest, asyncIterable);
+  const data = fs.readFileSync(dest, 'utf-8');
+  assert.deepStrictEqual(data, asyncIterable.expected);
+}
+
+async function doWriteInvalidValues() {
+  await Promise.all(
+    [42, 42n, {}, Symbol('42'), true, undefined, null, NaN].map((value) =>
+      assert.rejects(fsPromises.writeFile(dest, value), {
+        code: 'ERR_INVALID_ARG_TYPE',
+      })
+    )
+  );
+}
+
 async function doWriteWithCancel() {
   const controller = new AbortController();
   const { signal } = controller;
@@ -50,9 +145,18 @@ async function doReadWithEncoding() {
   assert.deepStrictEqual(data, syncData);
 }
 
-doWrite()
-  .then(doWriteWithCancel)
-  .then(doAppend)
-  .then(doRead)
-  .then(doReadWithEncoding)
-  .then(common.mustCall());
+(async () => {
+  await doWrite();
+  await doWriteWithCancel();
+  await doAppend();
+  await doRead();
+  await doReadWithEncoding();
+  await doWriteStream();
+  await doWriteStreamWithCancel();
+  await doWriteIterable();
+  await doWriteInvalidIterable();
+  await doWriteIterableWithEncoding();
+  await doWriteBufferIterable();
+  await doWriteAsyncIterable();
+  await doWriteInvalidValues();
+})().then(common.mustCall());