Reject emitted promises on error

src/core/EventEmitter.ts

@@ -74,8 +74,18 @@ export class EventEmitter<EventTypes extends Record<string, (...args: any) => an
     : EventParameters<EventTypes, Event>
   > {
     return new Promise((resolve, reject) => {
-      // TODO: handle errors
-      this.once(event, resolve as any);
+      const errorEvent = 'error' as keyof EventTypes;
+      const onEvent = ((value: unknown) => {
+        if (event !== errorEvent) this.off(errorEvent, onError);
+        resolve(value as any);
+      }) as EventListener<EventTypes, Event>;
+      const onError = ((error: unknown) => {
+        this.off(event, onEvent);
+        reject(error);
+      }) as EventListener<EventTypes, keyof EventTypes>;
+
+      if (event !== errorEvent) this.once(errorEvent, onError);
+      this.once(event, onEvent);
     });
   }
 

src/lib/EventEmitter.ts

@@ -74,8 +74,18 @@ export class EventEmitter<EventTypes extends Record<string, (...args: any) => an
     : EventParameters<EventTypes, Event>
   > {
     return new Promise((resolve, reject) => {
-      // TODO: handle errors
-      this.once(event, resolve as any);
+      const errorEvent = 'error' as keyof EventTypes;
+      const onEvent = ((value: unknown) => {
+        if (event !== errorEvent) this.off(errorEvent, onError);
+        resolve(value as any);
+      }) as EventListener<EventTypes, Event>;
+      const onError = ((error: unknown) => {
+        this.off(event, onEvent);
+        reject(error);
+      }) as EventListener<EventTypes, keyof EventTypes>;
+
+      if (event !== errorEvent) this.once(errorEvent, onError);
+      this.once(event, onEvent);
     });
   }
 

tests/EventEmitter.test.ts

@@ -0,0 +1,82 @@
+import { EventEmitter as CoreEventEmitter } from 'openai/core/EventEmitter';
+import { EventEmitter as LibEventEmitter } from 'openai/lib/EventEmitter';
+
+type TestEvents = {
+  message: (message: string) => void;
+  error: (error: Error) => void;
+};
+
+interface TestEmitter {
+  emitted(event: keyof TestEvents): Promise<unknown>;
+  emit<Event extends keyof TestEvents>(event: Event, ...args: Parameters<TestEvents[Event]>): void;
+  hasListener(event: keyof TestEvents): boolean;
+}
+
+class CoreTestEmitter extends CoreEventEmitter<TestEvents> implements TestEmitter {
+  emit<Event extends keyof TestEvents>(event: Event, ...args: Parameters<TestEvents[Event]>) {
+    this._emit(event, ...args);
+  }
+
+  hasListener(event: keyof TestEvents) {
+    return this._hasListener(event);
+  }
+}
+
+class LibTestEmitter extends LibEventEmitter<TestEvents> implements TestEmitter {
+  emit<Event extends keyof TestEvents>(event: Event, ...args: Parameters<TestEvents[Event]>) {
+    this._emit(event, ...args);
+  }
+
+  hasListener(event: keyof TestEvents) {
+    return this._hasListener(event);
+  }
+}
+
+function runEventEmitterTests(name: string, makeEmitter: () => TestEmitter) {
+  describe(`${name} EventEmitter`, () => {
+    it('resolves emitted() when the requested event is emitted', async () => {
+      const emitter = makeEmitter();
+      const promise = emitter.emitted('message');
+
+      expect(emitter.hasListener('message')).toBe(true);
+      expect(emitter.hasListener('error')).toBe(true);
+
+      emitter.emit('message', 'hello');
+
+      await expect(promise).resolves.toBe('hello');
+      expect(emitter.hasListener('message')).toBe(false);
+      expect(emitter.hasListener('error')).toBe(false);
+    });
+
+    it('rejects emitted() if error is emitted first', async () => {
+      const emitter = makeEmitter();
+      const error = new Error('boom');
+      const promise = emitter.emitted('message');
+
+      expect(emitter.hasListener('message')).toBe(true);
+      expect(emitter.hasListener('error')).toBe(true);
+
+      emitter.emit('error', error);
+
+      await expect(promise).rejects.toBe(error);
+      expect(emitter.hasListener('message')).toBe(false);
+      expect(emitter.hasListener('error')).toBe(false);
+    });
+
+    it('resolves emitted() when waiting for error', async () => {
+      const emitter = makeEmitter();
+      const error = new Error('boom');
+      const promise = emitter.emitted('error');
+
+      expect(emitter.hasListener('error')).toBe(true);
+
+      emitter.emit('error', error);
+
+      await expect(promise).resolves.toBe(error);
+      expect(emitter.hasListener('error')).toBe(false);
+    });
+  });
+}
+
+runEventEmitterTests('core', () => new CoreTestEmitter());
+runEventEmitterTests('lib', () => new LibTestEmitter());