Skip to content
rachel/workerd
File

Blob: src/node/internal/events.ts

typescript1053 lines
1// Copyright (c) 2017-2022 Cloudflare, Inc.
2// Licensed under the Apache 2.0 license found in the LICENSE file or at:
3//     https://opensource.org/licenses/Apache-2.0
4//
5// Adapted from Deno and Node.js:
6// Copyright 2018-2022 the Deno authors. All rights reserved. MIT license.
7//
8// Adapted from Node.js. Copyright Joyent, Inc. and other Node contributors.
9//
10// Permission is hereby granted, free of charge, to any person obtaining a
11// copy of this software and associated documentation files (the
12// "Software"), to deal in the Software without restriction, including
13// without limitation the rights to use, copy, modify, merge, publish,
14// distribute, sublicense, and/or sell copies of the Software, and to permit
15// persons to whom the Software is furnished to do so, subject to the
16// following conditions:
17//
18// The above copyright notice and this permission notice shall be included
19// in all copies or substantial portions of the Software.
20//
21// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
22// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
23// MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN
24// NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
25// DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
26// OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE
27// USE OR OTHER DEALINGS IN THE SOFTWARE.
28/* eslint-disable */
29 
30import {
31  AbortError,
32  ERR_INVALID_ARG_TYPE,
33  ERR_INVALID_THIS,
34  ERR_OUT_OF_RANGE,
35  ERR_UNHANDLED_ERROR,
36} from 'node-internal:internal_errors';
37import type {
38  EventEmitterAsyncResource as _EventEmitterAsyncResource,
39  EventEmitter as _EventEmitter,
40} from 'node:events';
41import {
42  validateAbortSignal,
43  validateBoolean,
44  validateFunction,
45  validateObject,
46} from 'node-internal:validators';
47import { spliceOne } from 'node-internal:internal_utils';
48import { nextTick, emitWarning } from 'node-internal:internal_process';
49import { default as async_hooks } from 'node-internal:async_hooks';
50const { AsyncResource } = async_hooks;
51 
52import { inspect } from 'node-internal:internal_inspect';
53 
54const enableNodejsProcessV2 =
55  !!Cloudflare.compatibilityFlags['enable_nodejs_process_v2'];
56 
57const kRejection = Symbol.for('nodejs.rejection');
58const kCapture = Symbol('kCapture');
59const kErrorMonitor = Symbol('events.errorMonitor');
60const kMaxEventTargetListeners = Symbol('events.maxEventTargetListeners');
61const kMaxEventTargetListenersWarned = Symbol(
62  'events.maxEventTargetListenersWarned'
63);
64 
65export interface EventEmitterOptions {
66  captureRejections?: boolean;
67}
68 
69export type EventName = string | symbol | number;
70export type EventCallback = ((...args: any[]) => unknown) & {
71  listener?: EventCallback;
72};
73// @ts-expect-error TS2417 Unrelated error.
74export declare class EventEmitter extends _EventEmitter {
75  constructor(opts?: EventEmitterOptions);
76  listenerCount(eventName: EventName): number;
77  emit(eventName: EventName, ...args: any[]): boolean;
78  on(eventName: EventName, listener: EventCallback): this;
79  once(eventName: EventName, listener: EventCallback): this;
80  addEventListener(eventName: EventName, listener: EventCallback): this;
81  removeEventListener(eventName: EventName, listener: EventCallback): this;
82 
83  [kRejection](err: unknown, eventName: EventName, ...args: unknown[]): void;
84  [kCapture]: boolean;
85 
86  _events: undefined | Record<EventName, EventCallback[]>;
87  _eventsCount: number;
88  _maxListeners: undefined | number;
89}
90 
91type AsyncResource = typeof AsyncResource;
92 
93declare var EventTarget: Function;
94 
95export function EventEmitter(
96  this: EventEmitter,
97  opts?: EventEmitterOptions
98): EventEmitter {
99  EventEmitter.init.call(this, opts);
100  return this;
101}
102 
103class EventEmitterReferencingAsyncResource extends AsyncResource {
104  #eventEmitter: EventEmitter;
105  constructor(emitter: EventEmitter) {
106    super('');
107    this.#eventEmitter = emitter;
108  }
109 
110  get eventEmitter() {
111    if (this.#eventEmitter === undefined)
112      throw new ERR_INVALID_THIS('EventEmitterReferencingAsyncResource');
113    return this.#eventEmitter;
114  }
115}
116 
117export class EventEmitterAsyncResource
118  extends EventEmitter
119  implements _EventEmitterAsyncResource
120{
121  #asyncResource: EventEmitterReferencingAsyncResource;
122 
123  constructor(options?: EventEmitterOptions) {
124    super(options);
125    this.#asyncResource = new EventEmitterReferencingAsyncResource(this);
126  }
127 
128  // @ts-expect-error TS2416 Not assignable to base type
129  get asyncResource(): AsyncResource {
130    if (this.#asyncResource === undefined)
131      throw new ERR_INVALID_THIS('EventEmitterAsyncResource');
132    // @ts-expect-error TS2741 Prototype is missing from type.
133    return this.#asyncResource;
134  }
135 
136  override emit(event: string | symbol, ...args: any[]): boolean {
137    if (this.#asyncResource === undefined)
138      throw new ERR_INVALID_THIS('EventEmitterAsyncResource');
139    args.unshift(super.emit, this, event);
140    Reflect.apply(
141      this.#asyncResource.runInAsyncScope,
142      this.#asyncResource,
143      args
144    );
145    return true;
146  }
147}
148 
149export function addAbortListener(
150  signal: AbortSignal | undefined,
151  listener: any
152) {
153  if (signal === undefined) {
154    throw new ERR_INVALID_ARG_TYPE('signal', 'AbortSignal', signal);
155  }
156  validateAbortSignal(signal, 'signal');
157  validateFunction(listener, 'listener');
158 
159  let removeEventListener: Function;
160  if (signal.aborted) {
161    queueMicrotask(() => listener());
162  } else {
163    signal.addEventListener('abort', listener, { once: true });
164    removeEventListener = () => {
165      signal.removeEventListener('abort', listener);
166    };
167  }
168  return {
169    __proto__: null,
170    [Symbol.dispose]() {
171      removeEventListener?.();
172    },
173  };
174}
175 
176export default EventEmitter;
177 
178EventEmitter.on = on;
179EventEmitter.once = once;
180EventEmitter.getEventListeners = getEventListeners;
181EventEmitter.getMaxListeners = _getMaxListeners;
182EventEmitter.setMaxListeners = setMaxListeners;
183EventEmitter.listenerCount = listenerCount;
184EventEmitter.EventEmitter = EventEmitter;
185EventEmitter.usingDomains = false;
186EventEmitter.captureRejectionSymbol = kRejection;
187EventEmitter.errorMonitor = kErrorMonitor;
188EventEmitter.EventEmitterAsyncResource = EventEmitterAsyncResource;
189 
190export const captureRejectionSymbol = EventEmitter.captureRejectionSymbol;
191export const captureRejections = false;
192export const errorMonitor = EventEmitter.errorMonitor;
193export const getMaxListeners = _getMaxListeners;
194export const usingDomains = EventEmitter.usingDomains;
195export let defaultMaxListeners = 10;
196 
197Object.defineProperties(EventEmitter, {
198  captureRejections: {
199    get(this: EventEmitter) {
200      return EventEmitter.prototype[kCapture];
201    },
202    set(this: EventEmitter, value: unknown): void {
203      validateBoolean(value, 'EventEmitter.captureRejections');
204 
205      EventEmitter.prototype[kCapture] = value;
206    },
207    enumerable: true,
208  },
209  defaultMaxListeners: {
210    enumerable: true,
211    get: function (this: EventEmitter): number {
212      return defaultMaxListeners;
213    },
214    set: function (this: EventEmitter, arg: unknown): void {
215      if (typeof arg !== 'number' || arg < 0 || Number.isNaN(arg)) {
216        throw new ERR_OUT_OF_RANGE(
217          'defaultMaxListeners',
218          'a non-negative number',
219          arg
220        );
221      }
222      defaultMaxListeners = arg;
223    },
224  },
225  kMaxEventTargetListeners: {
226    value: kMaxEventTargetListeners,
227    enumerable: false,
228    configurable: false,
229    writable: false,
230  },
231  kMaxEventTargetListenersWarned: {
232    value: kMaxEventTargetListenersWarned,
233    enumerable: false,
234    configurable: false,
235    writable: false,
236  },
237});
238 
239// The default for captureRejections is false
240Object.defineProperty(EventEmitter.prototype, kCapture, {
241  value: false,
242  writable: true,
243  enumerable: false,
244});
245 
246EventEmitter.init = function (this: EventEmitter, opts?: EventEmitterOptions) {
247  if (
248    this._events === undefined ||
249    this._events === Object.getPrototypeOf(this)._events
250  ) {
251    this._events = Object.create(null);
252    this._eventsCount = 0;
253  }
254 
255  (this as any)._maxListeners ??= undefined;
256  (this as any).domain = null;
257 
258  if (opts?.captureRejections) {
259    validateBoolean(opts.captureRejections, 'options.captureRejections');
260    (this as any)[kCapture] = Boolean(opts.captureRejections);
261  } else {
262    // Assigning the kCapture property directly saves an expensive
263    // prototype lookup in a very sensitive hot path.
264    (this as any)[kCapture] = EventEmitter.prototype[kCapture];
265  }
266};
267 
268export function setMaxListeners(
269  n = defaultMaxListeners,
270  ...eventTargets: any[]
271) {
272  if (typeof n !== 'number' || n < 0 || Number.isNaN(n)) {
273    throw new ERR_OUT_OF_RANGE('n', 'a non-negative number', n);
274  }
275  if (eventTargets.length === 0) {
276    defaultMaxListeners = n;
277  } else {
278    for (let i = 0; i < eventTargets.length; i++) {
279      const target = eventTargets[i];
280      if (target instanceof EventTarget) {
281        (target as any)[kMaxEventTargetListeners] = n;
282        (target as any)[kMaxEventTargetListenersWarned] = false;
283      } else if (typeof target.setMaxListeners === 'function') {
284        target.setMaxListeners(n);
285      } else {
286        throw new ERR_INVALID_ARG_TYPE(
287          'eventTargets',
288          ['EventEmitter', 'EventTarget'],
289          target
290        );
291      }
292    }
293  }
294}
295 
296EventEmitter.prototype._events = undefined;
297EventEmitter.prototype._eventsCount = 0;
298EventEmitter.prototype._maxListeners = undefined;
299EventEmitter.addAbortListener = addAbortListener;
300 
301function addCatch(
302  that: any,
303  promise: Promise<unknown>,
304  type: string | symbol,
305  args: any[]
306) {
307  if (!that[kCapture]) {
308    return;
309  }
310 
311  // Handle Promises/A+ spec, then could be a getter
312  // that throws on second use.
313  try {
314    const then = promise.then;
315 
316    if (typeof then === 'function') {
317      then.call(promise, undefined, function (err) {
318        // The callback is called with nextTick to avoid a follow-up
319        // rejection from this promise.
320        nextTick(emitUnhandledRejectionOrErr, that, err, type, args);
321      });
322    }
323  } catch (err) {
324    that.emit('error', err);
325  }
326}
327 
328function emitUnhandledRejectionOrErr(
329  ee: any,
330  err: any,
331  type: string | symbol,
332  args: any[]
333) {
334  if (typeof ee[kRejection] === 'function') {
335    ee[kRejection](err, type, ...args);
336  } else {
337    // We have to disable the capture rejections mechanism, otherwise
338    // we might end up in an infinite loop.
339    const prev = ee[kCapture];
340 
341    // If the error handler throws, it is not catchable and it
342    // will end up in 'uncaughtException'. We restore the previous
343    // value of kCapture in case the uncaughtException is present
344    // and the exception is handled.
345    try {
346      ee[kCapture] = false;
347      ee.emit('error', err);
348    } finally {
349      ee[kCapture] = prev;
350    }
351  }
352}
353 
354EventEmitter.prototype.setMaxListeners = function setMaxListeners(n: number) {
355  if (typeof n !== 'number' || n < 0 || Number.isNaN(n)) {
356    throw new ERR_OUT_OF_RANGE('n', 'a non-negative number', n);
357  }
358  this._maxListeners = n;
359  return this;
360};
361 
362function _getMaxListeners(that: any) {
363  if (that._maxListeners === undefined) {
364    return (EventEmitter as any).defaultMaxListeners;
365  }
366  return that._maxListeners;
367}
368 
369EventEmitter.prototype.getMaxListeners = function getMaxListeners(
370  this: EventEmitter
371): number {
372  return _getMaxListeners(this);
373};
374 
375EventEmitter.prototype.emit = function emit(
376  this: EventEmitter,
377  type: string | symbol,
378  ...args: any[]
379): boolean {
380  let doError = type === 'error';
381 
382  const events = this._events;
383  if (events !== undefined) {
384    if (doError && events[kErrorMonitor] !== undefined) {
385      this.emit(kErrorMonitor, ...args);
386    }
387    doError = doError && events.error === undefined;
388  } else if (!doError) {
389    return false;
390  }
391 
392  // If there is no 'error' event listener then throw.
393  if (doError) {
394    let er;
395    if (args.length > 0) {
396      er = args[0];
397    }
398    if (er instanceof Error) {
399      try {
400        const capture = {};
401        (Error as any).captureStackTrace(capture, EventEmitter.prototype.emit);
402      } catch {
403        // pass
404      }
405 
406      // Note: The comments on the `throw` lines are intentional, they show
407      // up in Node's output if this results in an unhandled exception.
408      throw er; // Unhandled 'error' event
409    }
410 
411    let stringifiedEr;
412    try {
413      stringifiedEr = inspect(er);
414    } catch {
415      stringifiedEr = er;
416    }
417 
418    // At least give some kind of context to the user
419    const err = new ERR_UNHANDLED_ERROR(stringifiedEr);
420    (err as any).context = er;
421    throw err; // Unhandled 'error' event
422  }
423 
424  const handler = events?.[type];
425 
426  if (handler === undefined) {
427    return false;
428  }
429 
430  if (typeof handler === 'function') {
431    const result = (handler as Function).apply(this, args);
432 
433    // We check if result is undefined first because that
434    // is the most common case so we do not pay any perf
435    // penalty
436    if (result !== undefined && result !== null) {
437      addCatch(this, result, type, args);
438    }
439  } else {
440    const len = handler.length;
441    const listeners = arrayClone(handler);
442    for (let i = 0; i < len; ++i) {
443      const result = listeners[i].apply(this, args);
444 
445      // We check if result is undefined first because that
446      // is the most common case so we do not pay any perf
447      // penalty.
448      // This code is duplicated because extracting it away
449      // would make it non-inlineable.
450      if (result !== undefined && result !== null) {
451        addCatch(this, result, type, args);
452      }
453    }
454  }
455 
456  return true;
457};
458 
459function _addListener(
460  target: any,
461  type: string | symbol,
462  listener: unknown,
463  prepend: boolean
464) {
465  let m;
466  let events;
467  let existing;
468 
469  validateFunction(listener, 'listener');
470 
471  events = target._events;
472  if (events === undefined) {
473    events = target._events = Object.create(null);
474    target._eventsCount = 0;
475  } else {
476    // To avoid recursion in the case that type === "newListener"! Before
477    // adding it to the listeners, first emit "newListener".
478    if (events.newListener !== undefined) {
479      target.emit('newListener', type, (listener as any).listener ?? listener);
480 
481      // Re-assign `events` because a newListener handler could have caused the
482      // this._events to be assigned to a new object
483      events = target._events;
484    }
485    existing = events[type];
486  }
487 
488  if (existing === undefined) {
489    // Optimize the case of one listener. Don't need the extra array object.
490    events[type] = listener;
491    ++target._eventsCount;
492  } else {
493    if (typeof existing === 'function') {
494      // Adding the second element, need to change to array.
495      existing = events[type] = prepend
496        ? [listener, existing]
497        : [existing, listener];
498      // If we've already got an array, just append.
499    } else if (prepend) {
500      existing.unshift(listener);
501    } else {
502      existing.push(listener);
503    }
504 
505    // Check for listener leak
506    m = _getMaxListeners(target);
507    if (m > 0 && existing.length > m && !existing.warned) {
508      existing.warned = true;
509      console.log(
510        'Possible EventEmitter memory leak detected. ' +
511          `${existing.length} ${String(type)} listeners ` +
512          `added to an EventEmitter. Use ` +
513          'emitter.setMaxListeners() to increase limit'
514      );
515      const w: EventEmitterError = Object.assign(
516        new Error(
517          'Possible EventEmitter memory leak detected. ' +
518            `${existing.length} ${String(type)} listeners ` +
519            `added to ${inspect(target, { depth: -1 })}. Use ` +
520            'emitter.setMaxListeners() to increase limit'
521        ),
522        {
523          name: 'MaxListenersExceededWarning',
524          emitter: target,
525          type: type,
526          count: existing.length,
527        }
528      );
529      // Only the newer process version compat adds process.emitWarning support.
530      if (enableNodejsProcessV2) emitWarning(w);
531    }
532  }
533 
534  return target;
535}
536 
537interface EventEmitterError extends Error {
538  name: string;
539  emitter: unknown;
540  type: string | symbol;
541  count: number;
542}
543 
544EventEmitter.prototype.addListener = function addListener(
545  this: EventEmitter,
546  type: string | symbol,
547  listener: unknown
548) {
549  return _addListener(this, type, listener, false);
550};
551 
552EventEmitter.prototype.on = EventEmitter.prototype.addListener;
553 
554EventEmitter.prototype.prependListener = function prependListener(
555  type: string | symbol,
556  listener: unknown
557) {
558  return _addListener(this, type, listener, true);
559};
560 
561function onceWrapper(this: any) {
562  if (!this.fired) {
563    this.target.removeListener(this.type, this.wrapFn);
564    this.fired = true;
565    if (arguments.length === 0) {
566      return this.listener.call(this.target);
567    }
568    return this.listener.apply(this.target, arguments);
569  }
570}
571 
572function _onceWrap(target: any, type: string | symbol, listener: unknown) {
573  const state = { fired: false, wrapFn: undefined, target, type, listener };
574  const wrapped = onceWrapper.bind(state);
575  (wrapped as any).listener = listener;
576  (state as any).wrapFn = wrapped;
577  return wrapped;
578}
579 
580EventEmitter.prototype.once = function once(
581  type: string | symbol,
582  listener: unknown
583) {
584  validateFunction(listener, 'listener');
585 
586  this.on(type, _onceWrap(this, type, listener));
587  return this;
588};
589 
590EventEmitter.prototype.prependOnceListener = function prependOnceListener(
591  type: string | symbol,
592  listener: unknown
593) {
594  validateFunction(listener, 'listener');
595 
596  this.prependListener(type, _onceWrap(this, type, listener));
597  return this;
598};
599 
600EventEmitter.prototype.removeListener = function removeListener(
601  type: string | symbol,
602  listener: unknown
603) {
604  validateFunction(listener, 'listener');
605 
606  const events = this._events;
607  if (events === undefined) {
608    return this;
609  }
610 
611  const list = events[type] as EventCallback | EventCallback[] | undefined;
612  if (list === undefined) {
613    return this;
614  }
615 
616  if (list === listener || ('listener' in list && list.listener === listener)) {
617    if (--this._eventsCount === 0) {
618      this._events = Object.create(null);
619    } else {
620      delete events[type];
621      if (events.removeListener) {
622        this.emit('removeListener', type, list.listener || listener);
623      }
624    }
625  } else if (typeof list !== 'function') {
626    let position = -1;
627 
628    for (let i = list.length - 1; i >= 0; i--) {
629      if (
630        list[i] === listener ||
631        (list[i] as EventCallback).listener === listener
632      ) {
633        position = i;
634        break;
635      }
636    }
637 
638    if (position < 0) {
639      return this;
640    }
641 
642    if (position === 0) {
643      list.shift();
644    } else {
645      spliceOne(list, position);
646    }
647 
648    if (list.length === 1) {
649      events[type] = list.at(0) as unknown as EventCallback[];
650    }
651 
652    if (events.removeListener !== undefined) {
653      this.emit('removeListener', type, listener);
654    }
655  }
656 
657  return this;
658};
659 
660EventEmitter.prototype.off = EventEmitter.prototype.removeListener;
661 
662EventEmitter.prototype.removeAllListeners = function removeAllListeners(
663  type: string | symbol
664) {
665  const events = this._events;
666  if (events === undefined) {
667    return this;
668  }
669 
670  // Not listening for removeListener, no need to emit
671  if (events.removeListener === undefined) {
672    if (arguments.length === 0) {
673      this._events = Object.create(null);
674      this._eventsCount = 0;
675    } else if (events[type] !== undefined) {
676      if (--this._eventsCount === 0) {
677        this._events = Object.create(null);
678      } else {
679        delete events[type];
680      }
681    }
682    return this;
683  }
684 
685  // Emit removeListener for all listeners on all events
686  if (arguments.length === 0) {
687    for (const key of Reflect.ownKeys(events)) {
688      if (key === 'removeListener') continue;
689      this.removeAllListeners(key);
690    }
691    this.removeAllListeners('removeListener');
692    this._events = Object.create(null);
693    this._eventsCount = 0;
694    return this;
695  }
696 
697  const listeners = events[type];
698 
699  if (typeof listeners === 'function') {
700    this.removeListener(type, listeners);
701  } else if (listeners !== undefined) {
702    // LIFO order
703    for (let i = listeners.length - 1; i >= 0; i--) {
704      this.removeListener(type, listeners[i] as EventListener);
705    }
706  }
707 
708  return this;
709};
710 
711function _listeners(target: any, type: string | symbol, unwrap: boolean) {
712  const events = target._events;
713 
714  if (events === undefined) {
715    return [];
716  }
717 
718  const evlistener = events[type];
719  if (evlistener === undefined) {
720    return [];
721  }
722 
723  if (typeof evlistener === 'function') {
724    return unwrap ? [evlistener.listener || evlistener] : [evlistener];
725  }
726 
727  return unwrap ? unwrapListeners(evlistener) : arrayClone(evlistener);
728}
729 
730EventEmitter.prototype.listeners = function listeners(type: string | symbol) {
731  return _listeners(this, type, true);
732};
733 
734EventEmitter.prototype.rawListeners = function rawListeners(
735  type: string | symbol
736) {
737  return _listeners(this, type, false);
738};
739 
740const _listenerCount = function listenerCount(
741  this: any,
742  type: string | symbol
743) {
744  const events = this._events;
745 
746  if (events !== undefined) {
747    const evlistener = events[type];
748 
749    if (typeof evlistener === 'function') {
750      return 1;
751    } else if (evlistener !== undefined) {
752      return evlistener.length;
753    }
754  }
755 
756  return 0;
757};
758 
759EventEmitter.prototype.listenerCount = _listenerCount;
760 
761export function listenerCount(emitter: any, type: string | symbol) {
762  if (typeof emitter.listenerCount === 'function') {
763    return emitter.listenerCount(type);
764  }
765  return _listenerCount.call(emitter, type);
766}
767 
768EventEmitter.prototype.eventNames = function eventNames() {
769  return this._eventsCount > 0 ? Reflect.ownKeys(this._events || {}) : [];
770};
771 
772function arrayClone(arr: any[]) {
773  // At least since V8 8.3, this implementation is faster than the previous
774  // which always used a simple for-loop
775  switch (arr.length) {
776    case 2:
777      return [arr[0], arr[1]];
778    case 3:
779      return [arr[0], arr[1], arr[2]];
780    case 4:
781      return [arr[0], arr[1], arr[2], arr[3]];
782    case 5:
783      return [arr[0], arr[1], arr[2], arr[3], arr[4]];
784    case 6:
785      return [arr[0], arr[1], arr[2], arr[3], arr[4], arr[5]];
786  }
787  return arr.slice();
788}
789 
790function unwrapListeners(arr: any[]) {
791  const ret = arrayClone(arr);
792  for (let i = 0; i < ret.length; ++i) {
793    const orig = ret[i].listener;
794    if (typeof orig === 'function') {
795      ret[i] = orig;
796    }
797  }
798  return ret;
799}
800 
801export function getEventListeners(emitterOrTarget: any, type: string | symbol) {
802  // First check if EventEmitter
803  if (typeof emitterOrTarget.listeners === 'function') {
804    return emitterOrTarget.listeners(type);
805  }
806  if (emitterOrTarget instanceof EventTarget) {
807    // Workers does not implement the ability to get the event listeners on an
808    // EventTarget the way that Node.js does. We simply return empty here.
809    return [];
810  }
811  throw new ERR_INVALID_ARG_TYPE(
812    'emitter',
813    ['EventEmitter', 'EventTarget'],
814    emitterOrTarget
815  );
816}
817 
818export interface OnceOptions {
819  signal?: AbortSignal;
820}
821 
822export async function once(
823  emitter: any,
824  name: string | symbol,
825  options: OnceOptions = {}
826) {
827  validateObject(options, 'options');
828  const { signal } = options;
829  validateAbortSignal(signal, 'options.signal');
830  if (signal?.aborted) {
831    throw new AbortError(undefined, { cause: signal.reason });
832  }
833  return new Promise((resolve, reject) => {
834    const errorListener = (err: any) => {
835      emitter.removeListener(name, resolver);
836      if (signal != null) {
837        eventTargetAgnosticRemoveListener(signal, 'abort', abortListener);
838      }
839      reject(err);
840    };
841    const resolver = (...args: any[]) => {
842      if (typeof emitter.removeListener === 'function') {
843        emitter.removeListener('error', errorListener);
844      }
845      if (signal != null) {
846        eventTargetAgnosticRemoveListener(signal, 'abort', abortListener);
847      }
848      resolve(args);
849    };
850    eventTargetAgnosticAddListener(emitter, name, resolver, { once: true });
851    if (name !== 'error' && typeof emitter.once === 'function') {
852      // EventTarget does not have `error` event semantics like Node
853      // EventEmitters, we listen to `error` events only on EventEmitters.
854      emitter.once('error', errorListener);
855    }
856    function abortListener() {
857      eventTargetAgnosticRemoveListener(emitter, name, resolver);
858      eventTargetAgnosticRemoveListener(emitter, 'error', errorListener);
859      reject(new AbortError());
860    }
861    if (signal != null) {
862      eventTargetAgnosticAddListener(signal, 'abort', abortListener, {
863        once: true,
864      });
865    }
866  });
867}
868 
869const AsyncIteratorPrototype = Object.getPrototypeOf(
870  Object.getPrototypeOf(async function* () {}).prototype
871);
872 
873function createIterResult(value: any, done: boolean) {
874  return { value, done };
875}
876 
877function eventTargetAgnosticRemoveListener(
878  emitter: any,
879  name: string | symbol,
880  listener: unknown,
881  flags: unknown = undefined
882) {
883  if (typeof emitter.removeListener === 'function') {
884    emitter.removeListener(name, listener);
885  } else if (typeof emitter.removeEventListener === 'function') {
886    emitter.removeEventListener(name, listener, flags);
887  } else {
888    throw new ERR_INVALID_ARG_TYPE('emitter', 'EventEmitter', emitter);
889  }
890}
891 
892interface AddListenerFlags {
893  once?: boolean;
894}
895 
896function eventTargetAgnosticAddListener(
897  emitter: any,
898  name: string | symbol,
899  listener: unknown,
900  flags: AddListenerFlags = {}
901) {
902  if (typeof emitter.on === 'function') {
903    if (flags?.once) {
904      emitter.once(name, listener);
905    } else {
906      emitter.on(name, listener);
907    }
908  } else if (typeof emitter.addEventListener === 'function') {
909    // EventTarget does not have `error` event semantics like Node
910    // EventEmitters, we do not listen to `error` events here.
911    emitter.addEventListener(
912      name,
913      (arg: unknown) => {
914        (listener as any)(arg);
915      },
916      flags
917    );
918  } else {
919    throw new ERR_INVALID_ARG_TYPE('emitter', 'EventEmitter', emitter);
920  }
921}
922 
923interface OnOptions {
924  signal?: AbortSignal;
925}
926 
927export function on(
928  emitter: any,
929  event: string | symbol,
930  options: OnOptions = {}
931) {
932  const signal = options?.signal;
933  validateAbortSignal(signal, 'options.signal');
934  if (signal?.aborted) {
935    throw new AbortError();
936  }
937 
938  const unconsumedEvents: any[] = [];
939  const unconsumedPromises: any[] = [];
940  let error: any = null;
941  let finished = false;
942 
943  const iterator = Object.setPrototypeOf(
944    {
945      next() {
946        // First, we consume all unread events
947        const value = unconsumedEvents.shift();
948        if (value) {
949          return Promise.resolve(createIterResult(value, false));
950        }
951 
952        // Then we error, if an error happened
953        // This happens one time if at all, because after 'error'
954        // we stop listening
955        if (error) {
956          const p = Promise.reject(error);
957          // Only the first element errors
958          error = null;
959          return p;
960        }
961 
962        // If the iterator is finished, resolve to done
963        if (finished) {
964          return Promise.resolve(createIterResult(undefined, true));
965        }
966 
967        // Wait until an event happens
968        return new Promise(function (resolve, reject) {
969          unconsumedPromises.push({ resolve, reject });
970        });
971      },
972 
973      return() {
974        eventTargetAgnosticRemoveListener(emitter, event, eventHandler);
975        eventTargetAgnosticRemoveListener(emitter, 'error', errorHandler);
976 
977        if (signal) {
978          eventTargetAgnosticRemoveListener(signal, 'abort', abortListener, {
979            once: true,
980          });
981        }
982 
983        finished = true;
984 
985        for (const promise of unconsumedPromises) {
986          promise.resolve(createIterResult(undefined, true));
987        }
988 
989        return Promise.resolve(createIterResult(undefined, true));
990      },
991 
992      throw(err: any) {
993        if (!err || !(err instanceof Error)) {
994          throw new ERR_INVALID_ARG_TYPE(
995            'EventEmitter.AsyncIterator',
996            'Error',
997            err
998          );
999        }
1000        error = err;
1001        eventTargetAgnosticRemoveListener(emitter, event, eventHandler);
1002        eventTargetAgnosticRemoveListener(emitter, 'error', errorHandler);
1003      },
1004 
1005      [Symbol.asyncIterator]() {
1006        return this;
1007      },
1008    },
1009    AsyncIteratorPrototype
1010  );
1011 
1012  eventTargetAgnosticAddListener(emitter, event, eventHandler);
1013  if (event !== 'error' && typeof emitter.on === 'function') {
1014    emitter.on('error', errorHandler);
1015  }
1016 
1017  if (signal) {
1018    eventTargetAgnosticAddListener(signal, 'abort', abortListener, {
1019      once: true,
1020    });
1021  }
1022 
1023  return iterator;
1024 
1025  function abortListener() {
1026    errorHandler(new AbortError());
1027  }
1028 
1029  function eventHandler(...args: any[]) {
1030    const promise = unconsumedPromises.shift();
1031    if (promise) {
1032      promise.resolve(createIterResult(args, false));
1033    } else {
1034      unconsumedEvents.push(args);
1035    }
1036  }
1037 
1038  function errorHandler(err: any) {
1039    finished = true;
1040 
1041    const toError = unconsumedPromises.shift();
1042 
1043    if (toError) {
1044      toError.reject(err);
1045    } else {
1046      // The next time we call next()
1047      error = err;
1048    }
1049 
1050    iterator.return();
1051  }
1052}