add maxQueue and overflow options

Author: pipobscure

lib/internal/fs/watchers.js

@@ -1,8 +1,11 @@
 'use strict';
 
 const {
+  ArrayPrototypePush,
+  ArrayPrototypeShift,
   FunctionPrototypeCall,
   ObjectDefineProperty,
+  ObjectPrototypeHasOwnProperty,
   ObjectSetPrototypeOf,
   PromiseWithResolvers,
   Symbol,
@@ -14,13 +17,13 @@ const {
   codes: {
     ERR_INVALID_ARG_VALUE,
   },
+  genericNodeError,
 } = require('internal/errors');
+
 const {
   kEmptyObject,
 } = require('internal/util');
 
-const FixedQueue = require('internal/fixed_queue');
-
 const {
   kFsStatsFieldsNumber,
   StatWatcher: _StatWatcher,
@@ -47,6 +50,8 @@ const {
   validateBoolean,
   validateObject,
   validateUint32,
+  validateInteger,
+  validateOneOf,
 } = require('internal/validators');
 
 const {
@@ -311,11 +316,15 @@ async function* watch(filename, options = kEmptyObject) {
     persistent = true,
     recursive = false,
     encoding = 'utf8',
+    maxQueue = 2048,
+    overflow = 'swallow',
     signal,
   } = options;
 
   validateBoolean(persistent, 'options.persistent');
   validateBoolean(recursive, 'options.recursive');
+  validateInteger(maxQueue, 'options.maxQueue');
+  validateOneOf(overflow, 'options.overflow', ['swallow', 'throw']);
   validateAbortSignal(signal, 'options.signal');
 
   if (encoding && !isEncoding(encoding)) {
@@ -328,7 +337,7 @@ async function* watch(filename, options = kEmptyObject) {
 
   const handle = new FSEvent();
   let { promise, resolve } = PromiseWithResolvers();
-  const queue = new FixedQueue();
+  const queue = [];
   const oncancel = () => {
     handle.close();
     resolve();
@@ -348,12 +357,21 @@ async function* watch(filename, options = kEmptyObject) {
         });
         error.filename = filename;
         handle.close();
-        queue.push(error);
+        ArrayPrototypePush(queue, error);
         resolve();
         return;
       }
-      queue.push({ eventType, filename });
-      resolve();
+      if (queue.length < maxQueue) {
+        ArrayPrototypePush(queue, { eventType, filename });
+        resolve();
+      } else if (overflow === 'throw') {
+        const overflowError = genericNodeError('fs.watch maxQueue exceeded');
+        queue.length = 0;
+        ArrayPrototypePush(queue, overflowError);
+        resolve();
+      } else {
+        process.emitWarning('fs.watch maxQueue exceeded');
+      }
     };
 
     const err = handle.start(path, persistent, recursive, encoding);
@@ -372,13 +390,18 @@ async function* watch(filename, options = kEmptyObject) {
 
     while (!signal?.aborted) {
       await promise;
-      while (!queue.isEmpty()) {
-        const item = queue.shift();
-        if (item instanceof UVException) throw item;
-        yield item;
+      while (queue.length) {
+        const item = ArrayPrototypeShift(queue);
+        if (
+          ObjectPrototypeHasOwnProperty(item, 'eventType') &&
+          ObjectPrototypeHasOwnProperty(item, 'filename')
+        ) {
+          yield item;
+        } else {
+          throw item;
+        }
       }
       ({ promise, resolve } = PromiseWithResolvers());
-
     }
     if (signal?.aborted) {
       throw new AbortError(undefined, { cause: signal?.reason });