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 <matteo.collina@gmail.com> Reviewed-By: Mattias Buelens <mattias@buelens.com> Reviewed-By: Luigi Pinca <luigipinca@gmail.com> Reviewed-By: James M Snell <jasnell@gmail.com>

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');
  }