feat(node:events): finish `EE.AsyncIterator` and `node:events.on()`, add some tests

4 files changed, 144 insertions, 240 deletions

diff --git a/src/bun.js/bindings/headers-cpp.h b/src/bun.js/bindings/headers-cpp.h
index 06a42eb57..5eac6785d 100644
--- a/src/bun.js/bindings/headers-cpp.h
+++ b/src/bun.js/bindings/headers-cpp.h
@@ -1,4 +1,4 @@
-//-- AUTOGENERATED FILE -- 1680036484
+//-- AUTOGENERATED FILE -- 1680635796
 // clang-format off
 #pragma once
 
diff --git a/src/bun.js/bindings/headers.h b/src/bun.js/bindings/headers.h
index a53938787..5b0e9527c 100644
--- a/src/bun.js/bindings/headers.h
+++ b/src/bun.js/bindings/headers.h
@@ -1,5 +1,5 @@
 // clang-format off
-//-- AUTOGENERATED FILE -- 1680036484
+//-- AUTOGENERATED FILE -- 1680635796
 #pragma once
 
 #include <stddef.h>
diff --git a/src/bun.js/builtins/cpp/NodeEventsBuiltins.cpp b/src/bun.js/builtins/cpp/NodeEventsBuiltins.cpp
index 6ae9863cb..c6f2b3895 100644
--- a/src/bun.js/builtins/cpp/NodeEventsBuiltins.cpp
+++ b/src/bun.js/builtins/cpp/NodeEventsBuiltins.cpp
@@ -51,42 +51,43 @@ namespace WebCore {
 const JSC::ConstructAbility s_nodeEventsOnAsyncIteratorCodeConstructAbility = JSC::ConstructAbility::CannotConstruct;
 const JSC::ConstructorKind s_nodeEventsOnAsyncIteratorCodeConstructorKind = JSC::ConstructorKind::None;
 const JSC::ImplementationVisibility s_nodeEventsOnAsyncIteratorCodeImplementationVisibility = JSC::ImplementationVisibility::Public;
-const int s_nodeEventsOnAsyncIteratorCodeLength = 4024;
+const int s_nodeEventsOnAsyncIteratorCodeLength = 4473;
 static const JSC::Intrinsic s_nodeEventsOnAsyncIteratorCodeIntrinsic = JSC::NoIntrinsic;
 const char* const s_nodeEventsOnAsyncIteratorCode =
     "(function (emitter, event, options) {\n" \
     "  \"use strict\";\n" \
     "\n" \
-    "  //\n" \
-    "  //\n" \
-    "  //\n" \
-    "  //\n" \
-    "  //\n" \
-    "  //\n" \
-    "  //\n" \
-    "  //\n" \
-    "  //\n" \
-    "  //\n" \
+    "  var { AbortSignal, Number, Error } = globalThis;\n" \
     "\n" \
-    "  //\n" \
-    "  var { Object, Number, console, Symbol } = globalThis;\n" \
-    "  //\n" \
+    "  var AbortError = class AbortError extends Error {\n" \
+    "    constructor(message = \"The operation was aborted\", options = void 0) {\n" \
+    "      if (options !== void 0 && typeof options !== \"object\") {\n" \
+    "        throw new Error(`Invalid AbortError options:\\n" \
+    "\\n" \
+    "${JSON.stringify(options, null, 2)}`);\n" \
+    "      }\n" \
+    "      super(message, options);\n" \
+    "      this.code = \"ABORT_ERR\";\n" \
+    "      this.name = \"AbortError\";\n" \
+    "    }\n" \
+    "  };\n" \
     "\n" \
-    "  function isUndefinedOrNull(value) {\n" \
-    "    return value === undefined || value === null;\n" \
-    "  }\n" \
+    "  if (@isUndefinedOrNull(emitter)) @throwTypeError(\"emitter is required\");\n" \
+    "  //\n" \
+    "  if (!(typeof emitter === \"object\" && @isCallable(emitter.emit) && @isCallable(emitter.on)))\n" \
+    "    @throwTypeError(\"emitter must be an EventEmitter\");\n" \
     "\n" \
-    "  if (isUndefinedOrNull(options)) options = {};\n" \
+    "  if (@isUndefinedOrNull(options)) options = {};\n" \
     "\n" \
     "  //\n" \
     "  var signal = options.signal;\n" \
-    "  //\n" \
-    "  //\n" \
+    "  if (!@isUndefinedOrNull(signal) && !(signal instanceof AbortSignal))\n" \
+    "    @throwTypeError(\"options.signal must be an AbortSignal\");\n" \
     "\n" \
-    "  //\n" \
-    "  //\n" \
-    "  //\n" \
-    "  //\n" \
+    "  if (signal?.aborted) {\n" \
+    "    //\n" \
+    "    throw new AbortError(@undefined, { cause: signal?.reason });\n" \
+    "  }\n" \
     "\n" \
     "  var highWatermark = options.highWatermark ?? Number.MAX_SAFE_INTEGER;\n" \
     "  if (highWatermark < 1) \n" \
@@ -105,9 +106,9 @@ const char* const s_nodeEventsOnAsyncIteratorCode =
     "  var size = 0;\n" \
     "  var listeners = [];\n" \
     "\n" \
-    "  //\n" \
-    "  //\n" \
-    "  //\n" \
+    "  function abortListener() {\n" \
+    "    errorHandler(new AbortError(@undefined, { cause: signal?.reason }));\n" \
+    "  }\n" \
     "\n" \
     "  function eventHandler(value) {\n" \
     "    if (unconsumedPromises.isEmpty()) {\n" \
@@ -117,22 +118,22 @@ const char* const s_nodeEventsOnAsyncIteratorCode =
     "        emitter.pause();\n" \
     "      }\n" \
     "      unconsumedEvents.push(value);\n" \
-    "    } else unconsumedPromises.shift().@resolve.@call(undefined, [value]);\n" \
+    "    } else unconsumedPromises.shift().@resolve.@call(@undefined, [value]);\n" \
     "  }\n" \
     "\n" \
     "  function closeHandler() {\n" \
     "    removeAllListeners(listeners);\n" \
     "    finished = true;\n" \
     "    while (!unconsumedPromises.isEmpty()) {\n" \
-    "      unconsumedPromises.shift().@resolve.@call(undefined, [undefined]);\n" \
+    "      const promise = unconsumedPromises.shift();\n" \
+    "      promise.@resolve.@call(@undefined, [@undefined]);\n" \
     "    }\n" \
-    "  \n" \
-    "    return @createFulfilledPromise([undefined]);\n" \
+    "    return @createFulfilledPromise([@undefined]);\n" \
     "  }\n" \
     "\n" \
     "  function errorHandler(err) {\n" \
     "    if (unconsumedPromises.isEmpty()) error = err;\n" \
-    "    else unconsumedPromises.shift().@reject.@call(undefined, err);\n" \
+    "    else unconsumedPromises.shift().@reject.@call(@undefined, err);\n" \
     "  \n" \
     "    closeHandler();\n" \
     "  }\n" \
@@ -140,37 +141,30 @@ const char* const s_nodeEventsOnAsyncIteratorCode =
     "  function addEventListener(emitter, event, handler) {\n" \
     "    emitter.on(event, handler);\n" \
     "    listeners.push([emitter, event, handler]);\n" \
-    "    //\n" \
     "  }\n" \
     "  \n" \
     "  function removeAllListeners() {\n" \
-    "    var heldEmitter;\n" \
     "    while (listeners.length > 0) {\n" \
-    "      //\n" \
     "      var entry = listeners.pop();\n" \
     "      var [emitter, event, handler] = entry;\n" \
-    "      if (event === \"foo\") heldEmitter = emitter;\n" \
-    "      console.log(emitter, event, handler);\n" \
     "      emitter.off(event, handler);\n" \
     "    }\n" \
-    "    console.log(heldEmitter.listenerCount(\"foo\"));\n" \
     "  }\n" \
     "\n" \
-    "  var iterator = async function* NodeEventsOnAsyncIterator() {\n" \
-    "      //\n" \
-    "      while (!finished) {\n" \
-    "        if (size) {\n" \
-    "          var values = [];\n" \
-    "          while (size) {\n" \
-    "            values.push(unconsumedEvents.shift());\n" \
-    "            size--;\n" \
-    "            if (paused && size < lowWatermark) {\n" \
-    "              emitter.resume();\n" \
-    "              paused = false;\n" \
-    "              break;\n" \
-    "            }\n" \
+    "  var createIterator = async function* NodeEventsOnAsyncIterator() {\n" \
+    "    //\n" \
+    "    try {\n" \
+    "      while (true) {\n" \
+    "        //\n" \
+    "        while (size) {\n" \
+    "          const value = unconsumedEvents.shift();\n" \
+    "          size--;\n" \
+    "          if (paused && size < lowWatermark) {\n" \
+    "            emitter.resume();\n" \
+    "            paused = false;\n" \
+    "            break;\n" \
     "          }\n" \
-    "          yield @createFulfilledPromise(values);\n" \
+    "          yield @createFulfilledPromise([value]);\n" \
     "        }\n" \
     "\n" \
     "        //\n" \
@@ -181,67 +175,19 @@ const char* const s_nodeEventsOnAsyncIteratorCode =
     "        }\n" \
     "\n" \
     "        //\n" \
-    "        if (finished) {\n" \
-    "          console.log(\"FINISHED\")\n" \
-    "          yield closeHandler();\n" \
-    "          break;\n" \
-    "        };\n" \
+    "        if (finished) break;\n" \
     "\n" \
     "        //\n" \
     "        var nextEventPromiseCapability = @newPromiseCapability(@Promise);\n" \
     "        unconsumedPromises.push(nextEventPromiseCapability);\n" \
     "        yield nextEventPromiseCapability.@promise;\n" \
     "      }\n" \
+    "    } finally {\n" \
+    "      closeHandler();\n" \
+    "    }\n" \
     "  };\n" \
     "\n" \
     "  //\n" \
-    "  //\n" \
-    "  //\n" \
-    "  //\n" \
-    "  //\n" \
-    "  //\n" \
-    "  //\n" \
-    "  //\n" \
-    "  //\n" \
-    "  //\n" \
-    "  //\n" \
-    "  //\n" \
-    "  //\n" \
-    "  //\n" \
-    "  //\n" \
-    "  //\n" \
-    "  //\n" \
-    "  //\n" \
-    "  //\n" \
-    "  //\n" \
-    "  //\n" \
-    "  //\n" \
-    "  //\n" \
-    "  //\n" \
-    "  //\n" \
-    "  //\n" \
-    "  //\n" \
-    "  //\n" \
-    "  //\n" \
-    "  //\n" \
-    "  //\n" \
-    "  //\n" \
-    "  //\n" \
-    "  //\n" \
-    "  //\n" \
-    "  //\n" \
-    "  //\n" \
-    "  //\n" \
-    "  //\n" \
-    "  //\n" \
-    "  //\n" \
-    "  //\n" \
-    "  //\n" \
-    "  //\n" \
-    "  //\n" \
-    "  //\n" \
-    "\n" \
-    "  //\n" \
     "  addEventListener(emitter, event, eventHandler);\n" \
     "  if (event !== \"error\" && typeof emitter.on === \"function\") {\n" \
     "    addEventListener(emitter, \"error\", errorHandler);\n" \
@@ -253,23 +199,30 @@ const char* const s_nodeEventsOnAsyncIteratorCode =
     "    }\n" \
     "  }\n" \
     "\n" \
-    "  //\n" \
-    "  //\n" \
+    "  if (signal)\n" \
+    "    signal.once(\"abort\", abortListener);\n" \
     "\n" \
-    "  return {\n" \
-    "    throw: (err) => {\n" \
-    "      if (err === undefined || err === null || !(err instanceof Error)) {\n" \
-    "        @throwTypeError(\"The argument must be an instance of Error\");\n" \
-    "      }\n" \
-    "      errorHandler(err);\n" \
+    "  var iterator = createIterator();\n" \
+    "\n" \
+    "  @Object.defineProperties(iterator, {\n" \
+    "    return: {\n" \
+    "      value: function() {\n" \
+    "        return closeHandler();\n" \
+    "      },\n" \
     "    },\n" \
-    "    return: () => {\n" \
-    "      console.log(\"we're here\");\n" \
-    "      return closeHandler();\n" \
+    "    throw: {\n" \
+    "      value: function(err) {\n" \
+    "        if (!err || !(err instanceof Error)) {\n" \
+    "          throw new TypeError(\"EventEmitter.AsyncIterator must be called with an error\");\n" \
+    "        }\n" \
+    "        errorHandler(err);\n" \
+    "      },\n" \
     "    },\n" \
-    "    next: () => iterator.next(),\n" \
-    "    [Symbol.asyncIterator]: iterator,\n" \
-    "  };\n" \
+    "    [Symbol.asyncIterator]: {\n" \
+    "      value: function() { return this; }\n" \
+    "    },\n" \
+    "  });\n" \
+    "  return iterator;\n" \
     "})\n" \
 ;
 
diff --git a/src/bun.js/builtins/js/NodeEvents.js b/src/bun.js/builtins/js/NodeEvents.js
index 02e944fde..6c80f81c8 100644
--- a/src/bun.js/builtins/js/NodeEvents.js
+++ b/src/bun.js/builtins/js/NodeEvents.js
@@ -26,36 +26,35 @@
 function onAsyncIterator(emitter, event, options) {
   "use strict";
 
-  // var AbortError = class AbortError extends Error {
-  //   constructor(message = "The operation was aborted", options = void 0) {
-  //     if (options !== void 0 && typeof options !== "object") {
-  //       throw new Error(`Invalid AbortError options:\n\n${JSON.stringify(options, null, 2)}`);
-  //     }
-  //     super(message, options);
-  //     this.code = "ABORT_ERR";
-  //     this.name = "AbortError";
-  //   }
-  // };
-
-  // var { AbortSignal, Object, Number, console } = globalThis;
-  var { Object, Number, console, Symbol } = globalThis;
-  // console.log("AbortSignal", AbortSignal);
-
-  function isUndefinedOrNull(value) {
-    return value === undefined || value === null;
-  }
+  var { AbortSignal, Number, Error } = globalThis;
+
+  var AbortError = class AbortError extends Error {
+    constructor(message = "The operation was aborted", options = void 0) {
+      if (options !== void 0 && typeof options !== "object") {
+        throw new Error(`Invalid AbortError options:\n\n${JSON.stringify(options, null, 2)}`);
+      }
+      super(message, options);
+      this.code = "ABORT_ERR";
+      this.name = "AbortError";
+    }
+  };
+
+  if (@isUndefinedOrNull(emitter)) @throwTypeError("emitter is required");
+  // TODO: Do a more accurate check
+  if (!(typeof emitter === "object" && @isCallable(emitter.emit) && @isCallable(emitter.on)))
+    @throwTypeError("emitter must be an EventEmitter");
 
-  if (isUndefinedOrNull(options)) options = {};
+  if (@isUndefinedOrNull(options)) options = {};
 
   // Parameters validation
   var signal = options.signal;
-  // if (!isUndefinedOrNull(signal) && !(signal instanceof AbortSignal))
-  //   @throwTypeError("options.signal must be an AbortSignal");
+  if (!@isUndefinedOrNull(signal) && !(signal instanceof AbortSignal))
+    @throwTypeError("options.signal must be an AbortSignal");
 
-  // if (signal?.aborted) {
-  //   // TODO: Make this a builtin
-  //   throw new AbortError(undefined, { cause: signal?.reason });
-  // }
+  if (signal?.aborted) {
+    // TODO: Make this a builtin
+    throw new AbortError(@undefined, { cause: signal?.reason });
+  }
 
   var highWatermark = options.highWatermark ?? Number.MAX_SAFE_INTEGER;
   if (highWatermark < 1) 
@@ -74,9 +73,9 @@ function onAsyncIterator(emitter, event, options) {
   var size = 0;
   var listeners = [];
 
-  // function abortListener() {
-  //   errorHandler(new AbortError(undefined, { cause: signal?.reason }));
-  // }
+  function abortListener() {
+    errorHandler(new AbortError(@undefined, { cause: signal?.reason }));
+  }
 
   function eventHandler(value) {
     if (unconsumedPromises.isEmpty()) {
@@ -86,22 +85,22 @@ function onAsyncIterator(emitter, event, options) {
         emitter.pause();
       }
       unconsumedEvents.push(value);
-    } else unconsumedPromises.shift().@resolve.@call(undefined, [value]);
+    } else unconsumedPromises.shift().@resolve.@call(@undefined, [value]);
   }
 
   function closeHandler() {
     removeAllListeners(listeners);
     finished = true;
     while (!unconsumedPromises.isEmpty()) {
-      unconsumedPromises.shift().@resolve.@call(undefined, [undefined]);
+      const promise = unconsumedPromises.shift();
+      promise.@resolve.@call(@undefined, [@undefined]);
     }
-  
-    return @createFulfilledPromise([undefined]);
+    return @createFulfilledPromise([@undefined]);
   }
 
   function errorHandler(err) {
     if (unconsumedPromises.isEmpty()) error = err;
-    else unconsumedPromises.shift().@reject.@call(undefined, err);
+    else unconsumedPromises.shift().@reject.@call(@undefined, err);
   
     closeHandler();
   }
@@ -109,40 +108,33 @@ function onAsyncIterator(emitter, event, options) {
   function addEventListener(emitter, event, handler) {
     emitter.on(event, handler);
     listeners.push([emitter, event, handler]);
-    // @arrayPush(listeners, [emitter, event, handler]);
   }
   
   function removeAllListeners() {
-    var heldEmitter;
     while (listeners.length > 0) {
-      // var [emitter, event, handler] = @arrayPop(listeners);
       var entry = listeners.pop();
       var [emitter, event, handler] = entry;
-      if (event === "foo") heldEmitter = emitter;
-      console.log(emitter, event, handler);
       emitter.off(event, handler);
     }
-    console.log(heldEmitter.listenerCount("foo"));
   }
 
-  var iterator = async function* NodeEventsOnAsyncIterator() {
-      // First, we consume all unread events
-      while (!finished) {
-        if (size) {
-          var values = [];
-          while (size) {
-            values.push(unconsumedEvents.shift());
-            size--;
-            if (paused && size < lowWatermark) {
-              emitter.resume();
-              paused = false;
-              break;
-            }
+  var createIterator = async function* NodeEventsOnAsyncIterator() {
+    // First, we consume all unread events
+    try {
+      while (true) {
+        // Go through queued events
+        while (size) {
+          const value = unconsumedEvents.shift();
+          size--;
+          if (paused && size < lowWatermark) {
+            emitter.resume();
+            paused = false;
+            break;
           }
-          yield @createFulfilledPromise(values);
+          yield @createFulfilledPromise([value]);
         }
 
-        // Then we error, if an error happened
+        // Check if error happened before yielding anything
         // This happens one time if at all, because after 'error'
         // we stop listening
         if (error) {
@@ -150,66 +142,18 @@ function onAsyncIterator(emitter, event, options) {
         }
 
         // If the iterator is finished, break
-        if (finished) {
-          console.log("FINISHED")
-          yield closeHandler();
-          break;
-        };
+        if (finished) break;
 
         // Wait until an event happens
         var nextEventPromiseCapability = @newPromiseCapability(@Promise);
         unconsumedPromises.push(nextEventPromiseCapability);
         yield nextEventPromiseCapability.@promise;
       }
+    } finally {
+      closeHandler();
+    }
   };
 
-  // // TODO: Use builtin
-  // Object.defineProperties(iterator, {
-  //   "throw": {
-  //     value: (err) => {
-  //       // TODO: Use Error builtin?
-  //       if (err === undefined || err === null || !(err instanceof Error)) {
-  //         @throwTypeError("The argument must be an instance of Error");
-  //       }
-  //       errorHandler(err);
-  //     },
-  //   },
-  //   "return": {
-  //     value: () => {
-  //       return closeHandler();
-  //     }
-  //   },
-  //   [Symbol.asyncIterator]: {
-  //     value: () => iterator,
-  //   },
-  //   // [kWatermarkData]: {
-  //   //   /**
-  //   //    * The current queue size
-  //   //    */
-  //   //   get size() {
-  //   //     return size;
-  //   //   },
-  //   //   /**
-  //   //    * The low watermark. The emitter is resumed every time size is lower than it
-  //   //    */
-  //   //   get low() {
-  //   //     return lowWatermark;
-  //   //   },
-  //   //   /**
-  //   //    * The high watermark. The emitter is paused every time size is higher than it
-  //   //    */
-  //   //   get high() {
-  //   //     return highWatermark;
-  //   //   },
-  //   //   /**
-  //   //    * It checks whether the emitter is paused by the watermark controller or not
-  //   //    */
-  //   //   get isPaused() {
-  //   //     return paused;
-  //   //   },
-  //   // },
-  // });
-
   // Adding event handlers
   addEventListener(emitter, event, eventHandler);
   if (event !== "error" && typeof emitter.on === "function") {
@@ -222,21 +166,28 @@ function onAsyncIterator(emitter, event, options) {
     }
   }
 
-  // if (signal)
-  //   signal.once("abort", abortListener);
+  if (signal)
+    signal.once("abort", abortListener);
 
-  return {
-    throw: (err) => {
-      if (err === undefined || err === null || !(err instanceof Error)) {
-        @throwTypeError("The argument must be an instance of Error");
-      }
-      errorHandler(err);
+  var iterator = createIterator();
+
+  @Object.defineProperties(iterator, {
+    return: {
+      value: function() {
+        return closeHandler();
+      },
+    },
+    throw: {
+      value: function(err) {
+        if (!err || !(err instanceof Error)) {
+          throw new TypeError("EventEmitter.AsyncIterator must be called with an error");
+        }
+        errorHandler(err);
+      },
     },
-    return: () => {
-      console.log("we're here");
-      return closeHandler();
+    [Symbol.asyncIterator]: {
+      value: function() { return this; }
     },
-    next: () => iterator.next(),
-    [Symbol.asyncIterator]: iterator,
-  };
+  });
+  return iterator;
 }