[v14.x Backport] fs: writeFile support AsyncIterable, Iterable & Stream as data argument

doc/api/fs.md

@@ -495,6 +495,9 @@ the end of the file.
 <!-- YAML
 added: v10.0.0
 changes:
+  - version: REPLACEME
+    pr-url: https://github.com/nodejs/node/pull/37490
+    description: The `data` argument supports `AsyncIterable`, `Iterable` and `Stream`.
   - version: v14.12.0
     pr-url: https://github.com/nodejs/node/pull/34993
     description: The `data` parameter will stringify an object with an
@@ -505,14 +508,16 @@ changes:
                  strings anymore.
 -->
 
-* `data` {string|Buffer|Uint8Array|Object}
+* `data` {string|Buffer|Uint8Array|Object|AsyncIterable|Iterable
+  |Stream}
 * `options` {Object|string}
   * `encoding` {string|null} The expected character encoding when `data` is a
     string. **Default:** `'utf8'`
 * Returns: {Promise}
 
 Asynchronously writes data to a file, replacing the file if it already exists.
-`data` can be a string, a buffer, or an object with an own `toString` function
+`data` can be a string, a buffer, an {AsyncIterable} or {Iterable} object, or an
+object with an own `toString` function
 property. The promise is resolved with no arguments upon success.
 
 If `options` is a string, then it specifies the `encoding`.
@@ -1234,6 +1239,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` and `Stream`.
   - version: v14.17.0
     pr-url: https://github.com/nodejs/node/pull/35993
     description: The options argument may include an AbortSignal to abort an
@@ -1249,7 +1257,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

@@ -34,7 +34,7 @@ const {
 const binding = internalBinding('fs');
 const { Buffer } = require('buffer');
 
-const { codes, hideStackFrames } = require('internal/errors');
+const { AbortError, codes, hideStackFrames } = require('internal/errors');
 const {
   ERR_FS_FILE_TOO_LARGE,
   ERR_INVALID_ARG_TYPE,
@@ -73,6 +73,7 @@ const {
 const pathModule = require('path');
 const { promisify } = require('internal/util');
 const { watch } = require('internal/fs/watchers');
+const { isIterable } = require('internal/streams/utils');
 
 const kHandle = Symbol('kHandle');
 const kFd = Symbol('kFd');
@@ -254,8 +255,23 @@ async function fsCall(fn, handle, ...args) {
   }
 }
 
-async function writeFileHandle(filehandle, data, signal) {
-  // `data` could be any kind of typed array.
+function checkAborted(signal) {
+  if (signal && signal.aborted)
+    throw new AbortError();
+}
+
+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;
@@ -403,7 +419,7 @@ async function readv(handle, buffers, position) {
 }
 
 async function write(handle, buffer, offset, length, position) {
-  if (buffer.length === 0)
+  if (buffer && buffer.length === 0)
     return { bytesWritten: 0, buffer };
 
   if (isArrayBufferView(buffer)) {
@@ -645,22 +661,26 @@ 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);
 
   if (options.signal?.aborted) {
     throw lazyDOMException('The operation was aborted', 'AbortError');
   }
 
   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) {

lib/internal/streams/pipeline.js

@@ -7,7 +7,6 @@ const {
   ArrayIsArray,
   ReflectApply,
   SymbolAsyncIterator,
-  SymbolIterator,
 } = primordials;
 
 let eos;
@@ -22,6 +21,12 @@ const {
   ERR_STREAM_DESTROYED
 } = require('internal/errors').codes;
 
+const {
+  isIterable,
+  isReadable,
+  isStream,
+} = require('internal/streams/utils');
+
 let EE;
 let PassThrough;
 let Readable;
@@ -78,26 +83,6 @@ function popCallback(streams) {
   return streams.pop();
 }
 
-function isReadable(obj) {
-  return !!(obj && typeof obj.pipe === 'function');
-}
-
-function isWritable(obj) {
-  return !!(obj && typeof obj.write === 'function');
-}
-
-function isStream(obj) {
-  return isReadable(obj) || isWritable(obj);
-}
-
-function isIterable(obj, isAsync) {
-  if (!obj) return false;
-  if (isAsync === true) return typeof obj[SymbolAsyncIterator] === 'function';
-  if (isAsync === false) return typeof obj[SymbolIterator] === 'function';
-  return typeof obj[SymbolAsyncIterator] === 'function' ||
-    typeof obj[SymbolIterator] === 'function';
-}
-
 function makeAsyncIterable(val) {
   if (isIterable(val)) {
     return val;

lib/internal/streams/utils.js

@@ -0,0 +1,32 @@
+'use strict';
+
+const {
+  SymbolAsyncIterator,
+  SymbolIterator,
+} = primordials;
+
+function isReadable(obj) {
+  return !!(obj && typeof obj.pipe === 'function');
+}
+
+function isWritable(obj) {
+  return !!(obj && typeof obj.write === 'function');
+}
+
+function isStream(obj) {
+  return isReadable(obj) || isWritable(obj);
+}
+
+function isIterable(obj, isAsync) {
+  if (!obj) return false;
+  if (isAsync === true) return typeof obj[SymbolAsyncIterator] === 'function';
+  if (isAsync === false) return typeof obj[SymbolIterator] === 'function';
+  return typeof obj[SymbolAsyncIterator] === 'function' ||
+    typeof obj[SymbolIterator] === 'function';
+}
+
+module.exports = {
+  isIterable,
+  isReadable,
+  isStream,
+};

node.gyp

@@ -245,6 +245,7 @@
       'lib/internal/streams/state.js',
       'lib/internal/streams/pipeline.js',
       'lib/internal/streams/end-of-stream.js',
+      'lib/internal/streams/utils.js',
       'deps/v8/tools/splaytree.js',
       'deps/v8/tools/codemap.js',
       'deps/v8/tools/consarray.js',

test/parallel/test-bootstrap-modules.js

@@ -79,6 +79,7 @@ const expectedModules = new Set([
   'NativeModule internal/process/warning',
   'NativeModule internal/querystring',
   'NativeModule internal/source_map/source_map_cache',
+  'NativeModule internal/streams/utils',
   'NativeModule internal/timers',
   'NativeModule internal/url',
   'NativeModule internal/util',

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"/