stream: replace manual function validation with validateFunction

Replace repetitive manual function type checking with the existing
validateFunction in multiple stream operator functions.

PR-URL: #59529
Reviewed-By: Matteo Collina <[email protected]>
Reviewed-By: Mattias Buelens <[email protected]>
Reviewed-By: Luigi Pinca <[email protected]>
Reviewed-By: James M Snell <[email protected]>

Author: JinhyeokFang

lib/internal/streams/operators.js

@@ -18,7 +18,6 @@ const { AbortController, AbortSignal } = require('internal/abort_controller');
 const {
   AbortError,
   codes: {
-    ERR_INVALID_ARG_TYPE,
     ERR_INVALID_ARG_VALUE,
     ERR_MISSING_ARGS,
     ERR_OUT_OF_RANGE,
@@ -28,6 +27,7 @@ const {
   validateAbortSignal,
   validateInteger,
   validateObject,
+  validateFunction,
 } = require('internal/validators');
 const { kWeakHandler, kResistStopPropagation } = require('internal/event_target');
 const { finished } = require('internal/streams/end-of-stream');
@@ -66,10 +66,7 @@ function compose(stream, options) {
 }
 
 function map(fn, options) {
-  if (typeof fn !== 'function') {
-    throw new ERR_INVALID_ARG_TYPE(
-      'fn', ['Function', 'AsyncFunction'], fn);
-  }
+  validateFunction(fn, 'fn');
   if (options != null) {
     validateObject(options, 'options');
   }
@@ -223,10 +220,7 @@ async function some(fn, options = undefined) {
 }
 
 async function every(fn, options = undefined) {
-  if (typeof fn !== 'function') {
-    throw new ERR_INVALID_ARG_TYPE(
-      'fn', ['Function', 'AsyncFunction'], fn);
-  }
+  validateFunction(fn, 'fn');
   // https://en.wikipedia.org/wiki/De_Morgan%27s_laws
   return !(await some.call(this, async (...args) => {
     return !(await fn(...args));
@@ -241,10 +235,7 @@ async function find(fn, options) {
 }
 
 async function forEach(fn, options) {
-  if (typeof fn !== 'function') {
-    throw new ERR_INVALID_ARG_TYPE(
-      'fn', ['Function', 'AsyncFunction'], fn);
-  }
+  validateFunction(fn, 'fn');
   async function forEachFn(value, options) {
     await fn(value, options);
     return kEmpty;
@@ -254,10 +245,7 @@ async function forEach(fn, options) {
 }
 
 function filter(fn, options) {
-  if (typeof fn !== 'function') {
-    throw new ERR_INVALID_ARG_TYPE(
-      'fn', ['Function', 'AsyncFunction'], fn);
-  }
+  validateFunction(fn, 'fn');
   async function filterFn(value, options) {
     if (await fn(value, options)) {
       return value;
@@ -277,10 +265,7 @@ class ReduceAwareErrMissingArgs extends ERR_MISSING_ARGS {
 }
 
 async function reduce(reducer, initialValue, options) {
-  if (typeof reducer !== 'function') {
-    throw new ERR_INVALID_ARG_TYPE(
-      'reducer', ['Function', 'AsyncFunction'], reducer);
-  }
+  validateFunction(reducer, 'reducer');
   if (options != null) {
     validateObject(options, 'options');
   }