Formatting

Composed push iterator

See merge request GeneralProtocols/xo/utils!14

package-lock.json

@@ -11,6 +11,7 @@
             "dependencies": {
                 "@bitauth/libauth": "^3.1.0-next.8",
                 "@xo-cash/types": "0.0.3",
+                "eventemitter3": "^5.0.4",
                 "zod": "^4.3.6"
             },
             "devDependencies": {
@@ -1105,9 +1106,9 @@
             "license": "MIT"
         },
         "node_modules/@eslint/config-array/node_modules/brace-expansion": {
-            "version": "1.1.15",
+            "version": "1.1.16",
-            "resolved": "https://registry.npmjs.org/brace-expansion/-/brace-expansion-1.1.15.tgz",
+            "resolved": "https://registry.npmjs.org/brace-expansion/-/brace-expansion-1.1.16.tgz",
-            "integrity": "sha512-EwOCDEex4quD37XhqM3omwtMoJjr//isUZz1JopUNWms+4Z2ViyM/k1YIRePpoVNnQhENnxtFjLaxNHrT7xIUg==",
+            "integrity": "sha512-IDw48K2/2kRkg9LdJxurvq3lV3aBgq0REY89duEqFRthjlPdXHKMj7EnQOXVckxzgisinf3nHfrcE2FufFLXMw==",
             "dev": true,
             "license": "MIT",
             "dependencies": {
@@ -1186,9 +1187,9 @@
             "license": "MIT"
         },
         "node_modules/@eslint/eslintrc/node_modules/brace-expansion": {
-            "version": "1.1.15",
+            "version": "1.1.16",
-            "resolved": "https://registry.npmjs.org/brace-expansion/-/brace-expansion-1.1.15.tgz",
+            "resolved": "https://registry.npmjs.org/brace-expansion/-/brace-expansion-1.1.16.tgz",
-            "integrity": "sha512-EwOCDEex4quD37XhqM3omwtMoJjr//isUZz1JopUNWms+4Z2ViyM/k1YIRePpoVNnQhENnxtFjLaxNHrT7xIUg==",
+            "integrity": "sha512-IDw48K2/2kRkg9LdJxurvq3lV3aBgq0REY89duEqFRthjlPdXHKMj7EnQOXVckxzgisinf3nHfrcE2FufFLXMw==",
             "dev": true,
             "license": "MIT",
             "dependencies": {
@@ -3435,9 +3436,9 @@
             }
         },
         "node_modules/body-parser": {
-            "version": "1.20.5",
+            "version": "1.20.6",
-            "resolved": "https://registry.npmjs.org/body-parser/-/body-parser-1.20.5.tgz",
+            "resolved": "https://registry.npmjs.org/body-parser/-/body-parser-1.20.6.tgz",
-            "integrity": "sha512-3grm+/2tUOvu2cjJkvsIxrv/wVpfXQW4PsQHYm7yk4vfpu7Ekl6nEsYBoJUL6qDwZUx8wUhQ8tR2qz+ad9c9OA==",
+            "integrity": "sha512-p5tAzS57i5MV9fZFDj9LeIiTZEufbSe2eDozP+ElheSUq1m74CRq1jI4mYNDdVs9vQztXFLuk/Gd6BWTdwRJ5g==",
             "dev": true,
             "license": "MIT",
             "dependencies": {
@@ -3503,9 +3504,9 @@
             }
         },
         "node_modules/brace-expansion": {
-            "version": "5.0.6",
+            "version": "5.0.7",
-            "resolved": "https://registry.npmjs.org/brace-expansion/-/brace-expansion-5.0.6.tgz",
+            "resolved": "https://registry.npmjs.org/brace-expansion/-/brace-expansion-5.0.7.tgz",
-            "integrity": "sha512-kLpxurY4Z4r9sgMsyG0Z9uzsBlgiU/EFKhj/h91/8yHu0edo7XuixOIH3VcJ8kkxs6/jPzoI6U9Vj3WqbMQ94g==",
+            "integrity": "sha512-7oFy703dxfY3/NLxC1fh2SUCQ0H9rmAY+5EpDVfXjUTTs+HEwR2nYaqLv+GWcTsumwxPfiz6CzCNkwXwBUwqCA==",
             "dev": true,
             "license": "MIT",
             "dependencies": {
@@ -5292,9 +5293,9 @@
             "peer": true
         },
         "node_modules/eslint-plugin-import/node_modules/brace-expansion": {
-            "version": "1.1.15",
+            "version": "1.1.16",
-            "resolved": "https://registry.npmjs.org/brace-expansion/-/brace-expansion-1.1.15.tgz",
+            "resolved": "https://registry.npmjs.org/brace-expansion/-/brace-expansion-1.1.16.tgz",
-            "integrity": "sha512-EwOCDEex4quD37XhqM3omwtMoJjr//isUZz1JopUNWms+4Z2ViyM/k1YIRePpoVNnQhENnxtFjLaxNHrT7xIUg==",
+            "integrity": "sha512-IDw48K2/2kRkg9LdJxurvq3lV3aBgq0REY89duEqFRthjlPdXHKMj7EnQOXVckxzgisinf3nHfrcE2FufFLXMw==",
             "dev": true,
             "license": "MIT",
             "peer": true,
@@ -5379,9 +5380,9 @@
             "peer": true
         },
         "node_modules/eslint-plugin-jsx-a11y/node_modules/brace-expansion": {
-            "version": "1.1.15",
+            "version": "1.1.16",
-            "resolved": "https://registry.npmjs.org/brace-expansion/-/brace-expansion-1.1.15.tgz",
+            "resolved": "https://registry.npmjs.org/brace-expansion/-/brace-expansion-1.1.16.tgz",
-            "integrity": "sha512-EwOCDEex4quD37XhqM3omwtMoJjr//isUZz1JopUNWms+4Z2ViyM/k1YIRePpoVNnQhENnxtFjLaxNHrT7xIUg==",
+            "integrity": "sha512-IDw48K2/2kRkg9LdJxurvq3lV3aBgq0REY89duEqFRthjlPdXHKMj7EnQOXVckxzgisinf3nHfrcE2FufFLXMw==",
             "dev": true,
             "license": "MIT",
             "peer": true,
@@ -5468,9 +5469,9 @@
             "peer": true
         },
         "node_modules/eslint-plugin-react/node_modules/brace-expansion": {
-            "version": "1.1.15",
+            "version": "1.1.16",
-            "resolved": "https://registry.npmjs.org/brace-expansion/-/brace-expansion-1.1.15.tgz",
+            "resolved": "https://registry.npmjs.org/brace-expansion/-/brace-expansion-1.1.16.tgz",
-            "integrity": "sha512-EwOCDEex4quD37XhqM3omwtMoJjr//isUZz1JopUNWms+4Z2ViyM/k1YIRePpoVNnQhENnxtFjLaxNHrT7xIUg==",
+            "integrity": "sha512-IDw48K2/2kRkg9LdJxurvq3lV3aBgq0REY89duEqFRthjlPdXHKMj7EnQOXVckxzgisinf3nHfrcE2FufFLXMw==",
             "dev": true,
             "license": "MIT",
             "peer": true,
@@ -5542,9 +5543,9 @@
             "license": "MIT"
         },
         "node_modules/eslint/node_modules/brace-expansion": {
-            "version": "1.1.15",
+            "version": "1.1.16",
-            "resolved": "https://registry.npmjs.org/brace-expansion/-/brace-expansion-1.1.15.tgz",
+            "resolved": "https://registry.npmjs.org/brace-expansion/-/brace-expansion-1.1.16.tgz",
-            "integrity": "sha512-EwOCDEex4quD37XhqM3omwtMoJjr//isUZz1JopUNWms+4Z2ViyM/k1YIRePpoVNnQhENnxtFjLaxNHrT7xIUg==",
+            "integrity": "sha512-IDw48K2/2kRkg9LdJxurvq3lV3aBgq0REY89duEqFRthjlPdXHKMj7EnQOXVckxzgisinf3nHfrcE2FufFLXMw==",
             "dev": true,
             "license": "MIT",
             "dependencies": {
@@ -5690,6 +5691,12 @@
                 "node": ">= 0.6"
             }
         },
+        "node_modules/eventemitter3": {
+            "version": "5.0.4",
+            "resolved": "https://registry.npmjs.org/eventemitter3/-/eventemitter3-5.0.4.tgz",
+            "integrity": "sha512-mlsTRyGaPBjPedk6Bvw+aqbsXDtoAyAzm5MO7JgU+yVRyMQ5O8bD4Kcci7BS85f93veegeCPkL8R4GLClnjLFw==",
+            "license": "MIT"
+        },
         "node_modules/execa": {
             "version": "7.2.0",
             "resolved": "https://registry.npmjs.org/execa/-/execa-7.2.0.tgz",
@@ -7129,9 +7136,9 @@
             "license": "MIT"
         },
         "node_modules/js-yaml": {
-            "version": "4.2.0",
+            "version": "4.3.0",
-            "resolved": "https://registry.npmjs.org/js-yaml/-/js-yaml-4.2.0.tgz",
+            "resolved": "https://registry.npmjs.org/js-yaml/-/js-yaml-4.3.0.tgz",
-            "integrity": "sha512-ePWsvanv0DWuDRsW8dnt+R4jQ31SCRCQ7hhNcPXZPsoBZiemuZNYGf7adZdqX2D86j6rvKp3RpCxVTSb8WQlOw==",
+            "integrity": "sha512-1td788aAnnZ5qs7V2QIRl1owjtYpbKt749Y3xauqQgwIIGF/xXWz1wMTEBx5O3LK3lXLVuqXPdPxj2BoFHaW9Q==",
             "dev": true,
             "funding": [
                 {
@@ -7524,9 +7531,9 @@
             }
         },
         "node_modules/linkify-it": {
-            "version": "5.0.1",
+            "version": "5.0.2",
-            "resolved": "https://registry.npmjs.org/linkify-it/-/linkify-it-5.0.1.tgz",
+            "resolved": "https://registry.npmjs.org/linkify-it/-/linkify-it-5.0.2.tgz",
-            "integrity": "sha512-wVoTjP4Q6R0NW5hiZkVJaFZPWgtXfoGF+6LucL3/FtiNjmcHhYjEr5f1Kqjirc1nBW07J/ZuRFumqr2oqccEWg==",
+            "integrity": "sha512-ONTm2jCMAVZjgQa/Fy1kScXsuOoF5NPTsoFBdE1KVIZ2vAh/r9+Bqo+0jINCBYnavTPQZz38QzFTme79ENoN3Q==",
             "dev": true,
             "funding": [
                 {

package.json

@@ -45,6 +45,7 @@
     "dependencies": {
         "@bitauth/libauth": "^3.1.0-next.8",
         "@xo-cash/types": "0.0.3",
+        "eventemitter3": "^5.0.4",
         "zod": "^4.3.6"
     },
     "overrides": {

sandbox/sandbox-llm.ts

@@ -0,0 +1,93 @@
+import { SSESession } from '../source/sse-session/index.ts';
+
+// Because this is in a library, and we don't want to add the node types to this as it is intended to be used in a browser
+// we will just declare the process object here locally so we don't get type errors from this script
+declare const process: {
+    env: Record<string, string>;
+    stdout: {
+        write: (data: string) => void;
+    };
+};
+
+// Recommended URL: https://openrouter.ai/api/v1/chat/completions
+// Recommended Model: nvidia/nemotron-3-nano-omni-30b-a3b-reasoning:free
+// Exmaple Command: API_KEY="your-api-key" URL="https://openrouter.ai/api/v1/chat/completions" MODEL="ibm-granite/granite-4.1-8b" PROMPT="Hello, Tell me a joke about robots?" npx tsx ./sandbox/sandbox-llm.ts
+
+// Read the Environemt Variables for url, model, prompt and api key
+const url = process.env.URL ?? 'https://openrouter.ai/api/v1/chat/completions';
+const model = process.env.MODEL ?? 'nvidia/nemotron-3-nano-omni-30b-a3b-reasoning:free';
+const prompt = process.env.PROMPT ?? 'Hello, Tell me a joke about robots?';
+
+const apiKey = process.env.API_KEY ?? '';
+
+// Throw an error if the api key is not set
+if (!apiKey) {
+    throw new Error('API key is required');
+}
+
+// Create a function to get the auth header
+const getAuthHeader = (): string => {
+    return `Bearer ${apiKey}`;
+};
+
+// Create our sse session
+const sseSession = new SSESession(url, {
+    // LLMs use requests
+    method: 'POST',
+
+    // Create the body of the request
+    body: JSON.stringify({
+        model: model,
+        messages: [
+            {
+                role: 'user',
+                content: prompt,
+            },
+        ],
+        stream: true,
+    }),
+
+    // Create a custom request handler to set the headers
+    onRequest: async (requestInit): Promise<RequestInit> => {
+        requestInit.headers ??= {} as HeadersInit;
+
+        // Handle typescript annoyances
+        const headers = requestInit.headers as Record<string, string>;
+
+        // Set our headers - We could also do this using the `headers` property in the SSESession constructor
+        // Doing it here to demonstrate dynamic headers, for example a signed timestamp could be used to authenticate the request.
+        headers.Authorization = getAuthHeader();
+        headers['Content-Type'] = 'application/json';
+
+        return requestInit;
+    },
+});
+
+// Connect to the SSESession
+await sseSession.connect();
+
+// Loop over the message chunks using `for await`
+for await (const message of sseSession.messages) {
+    // Handle `[DONE]` (this may be specific to OpenRouter)
+    if (message.data === '[DONE]') {
+        continue;
+    }
+
+    // First, we will parse the event to JSON
+    const responseJson = JSON.parse(message.data);
+
+    // Then we will grab the relavent part of the response (we want to first grab the choices array)
+    const choices = responseJson.choices;
+
+    // Then we will grab the first choice
+    const firstChoice = choices[0];
+
+    // Then we will grab the next chunk of text
+    const messageContent = firstChoice.delta?.content;
+
+    // Then we will append the text to the console
+    process.stdout.write(messageContent || '');
+}
+
+// Just a terminal/node thing. If we dont put a new line, the console will overwrite the text with the `cwd` or next command input
+process.stdout.write('\n');

sandbox/sandbox-price-oracle.ts

@@ -0,0 +1,17 @@
+import { SSESession, type SSEvent } from '../source/sse-session/index.ts';
+
+// Command: npx tsx ./sandbox/sandbox-price-oracle.ts
+
+// Use the GP Price Oracle API
+const url = 'https://oracles.generalprotocols.com/sse/v1/messages';
+
+// Create our sse session
+const sseSession = await SSESession.create(url);
+
+// Create a message event handler
+sseSession.on('message', (message: SSEvent) => {
+    console.log(message);
+});
+
+// Connect to the SSESession
+await sseSession.connect();

source/event-emitter.ts

@@ -1,201 +0,0 @@
-export type EventMap = Record<string, unknown>;
-
-type Listener<T> = (detail: T) => void;
-
-/**
- * A listener entry.
- * @template T - The event type.
- */
-interface ListenerEntry<T> {
-    listener: Listener<T>;
-    wrappedListener: Listener<T>;
-    debounceTime?: number;
-    once?: boolean;
-}
-
-export type OffCallback = () => void;
-
-/**
- * A simple event emitter implementation.
- * @template T - The event map type.
- */
-export class EventEmitter<T extends EventMap> {
-    /**
-     * The listeners map.
-     * @private
-     */
-    private listeners: Map<keyof T, Set<ListenerEntry<T[keyof T]>>> = new Map();
-
-    /**
-     * Add a listener for an event.
-     * @param type - The event type.
-     * @param listener - The listener function.
-     * @param debounceMilliseconds - The debounce time in milliseconds.
-     * @returns An off callback that can be called to stop listening for events.
-     */
-    on<K extends keyof T>(type: K, listener: Listener<T[K]>, debounceMilliseconds?: number): OffCallback {
-        // Create a wrapped listener so that the debounce can be applied.
-        const wrappedListener = debounceMilliseconds && debounceMilliseconds > 0 ? this.debounce(listener, debounceMilliseconds) : listener;
-
-        // If the listeners map does not have the event type, create a new set.
-        if (!this.listeners.has(type)) {
-            this.listeners.set(type, new Set());
-        }
-
-        // Create a listener entry.
-        const listenerEntry: ListenerEntry<T[K]> = {
-            listener,
-            wrappedListener,
-            ...(debounceMilliseconds !== undefined ? { debounceTime: debounceMilliseconds } : {}),
-        };
-
-        // Add the listener entry to the listeners map.
-        this.listeners.get(type)?.add(listenerEntry as ListenerEntry<T[keyof T]>);
-
-        // Return an "off" callback that can be called to stop listening for events.
-        return () => this.off(type, listener);
-    }
-
-    /**
-     * Add a one-time listener for an event.
-     * @param type - The event type.
-     * @param listener - The listener function.
-     * @param debounceMilliseconds - The debounce time in milliseconds.
-     * @returns An off callback that can be called to stop listening for events.
-     */
-    once<K extends keyof T>(type: K, listener: Listener<T[K]>, debounceMilliseconds?: number): OffCallback {
-        const wrappedListener: Listener<T[K]> = (detail: T[K]) => {
-            this.off(type, listener);
-            listener(detail);
-        };
-
-        // Create a debounced listener.
-        const debouncedListener =
-            debounceMilliseconds && debounceMilliseconds > 0 ? this.debounce(wrappedListener, debounceMilliseconds) : wrappedListener;
-
-        // If the listeners map does not have the event type, create a new set.
-        if (!this.listeners.has(type)) {
-            this.listeners.set(type, new Set());
-        }
-
-        // Create a listener entry.
-        const listenerEntry: ListenerEntry<T[K]> = {
-            listener,
-            wrappedListener: debouncedListener,
-            once: true,
-            ...(debounceMilliseconds !== undefined ? { debounceTime: debounceMilliseconds } : {}),
-        };
-
-        // Add the listener entry to the listeners map.
-        this.listeners.get(type)?.add(listenerEntry as ListenerEntry<T[keyof T]>);
-
-        // Return an "off" callback that can be called to stop listening for events.
-        return () => this.off(type, listener);
-    }
-
-    /**
-     * Remove a listener for an event.
-     * @param type - The event type.
-     * @param listener - The listener function.
-     */
-    off<K extends keyof T>(type: K, listener: Listener<T[K]>): void {
-        // Get the listeners for the event type.
-        const listeners = this.listeners.get(type);
-        if (!listeners) return;
-
-        // Find the listener entry.
-        const listenerEntry = Array.from(listeners).find((entry) => entry.listener === listener || entry.wrappedListener === listener);
-
-        // If the listener entry is found, remove it from the listeners map.
-        if (listenerEntry) {
-            listeners.delete(listenerEntry);
-        }
-    }
-
-    /**
-     * Emit an event.
-     * @param type - The event type.
-     * @param payload - The event payload.
-     * @returns True if there are listeners for the event, false otherwise.
-     */
-    emit<K extends keyof T>(type: K, payload: T[K]): boolean {
-        // Get the listeners for the event type.
-        const listeners = this.listeners.get(type);
-        if (!listeners) return false;
-
-        // Emit the event to all listeners.
-        listeners.forEach((entry) => {
-            entry.wrappedListener(payload);
-        });
-
-        // Return true if there are listeners for the event, false otherwise.
-        return listeners.size > 0;
-    }
-
-    /**
-     * Remove all listeners.
-     */
-    removeAllListeners(): void {
-        this.listeners.clear();
-    }
-
-    /**
-     * Wait for an event to be emitted.
-     * @param type - The event type.
-     * @param predicate - The predicate function.
-     * @param timeoutMs - The timeout in milliseconds.
-     * @returns The event payload.
-     */
-    async waitFor<K extends keyof T>(type: K, predicate: (payload: T[K]) => boolean, timeoutMs?: number): Promise<T[K]> {
-        // Create a promise to wait for the event to be emitted.
-        return new Promise((resolve, reject) => {
-            let timeoutId: ReturnType<typeof setTimeout> | undefined;
-
-            // Create a listener function.
-            const listener = (payload: T[K]): void => {
-                if (predicate(payload)) {
-                    // Clean up
-                    this.off(type, listener);
-                    if (timeoutId !== undefined) {
-                        clearTimeout(timeoutId);
-                    }
-
-                    resolve(payload);
-                }
-            };
-
-            // Set up timeout if specified
-            if (timeoutMs !== undefined) {
-                timeoutId = setTimeout(() => {
-                    this.off(type, listener);
-                    reject(new Error(`Timeout waiting for event "${String(type)}"`));
-                }, timeoutMs);
-            }
-
-            // Add the listener to the listeners map.
-            this.on(type, listener);
-        });
-    }
-
-    /**
-     * Debounce a function.
-     * @param func - The function to debounce.
-     * @param wait - The wait time in milliseconds.
-     * @returns The debounced function.
-     */
-    private debounce<K extends keyof T>(func: Listener<T[K]>, wait: number): Listener<T[K]> {
-        // Create a timeout variable.
-        let timeout: ReturnType<typeof setTimeout>;
-
-        return (detail: T[K]) => {
-            // If a debounce timer is already pending, clear it before scheduling the next one.
-            if (timeout !== undefined) {
-                clearTimeout(timeout);
-            }
-
-            timeout = setTimeout(() => {
-                func(detail);
-            }, wait);
-        };
-    }
-}

source/extended-json.ts

@@ -74,3 +74,23 @@ export const extendedJsonReviver = (_propertyKey: string, value: unknown): unkno
     // If the value does not match either pattern, return the original value
     return value;
 };
+
+/**
+ * Serializes an object to a string using the {@link extendedJsonReplacer}.
+ *
+ * @param object The object to serialize.
+ * @returns The string representation of the object in Extended JSON format.
+ */
+export const toExtendedJson = (object: unknown): string => {
+    return JSON.stringify(object, extendedJsonReplacer);
+};
+
+/**
+ * Deserializes a string to an object using the {@link extendedJsonReviver}.
+ *
+ * @param serializedObject The string to deserialize.
+ * @returns The object reconstructed from the string.
+ */
+export const fromExtendedJson = (serializedObject: string): unknown => {
+    return JSON.parse(serializedObject, extendedJsonReviver);
+};

source/index.ts

@@ -1,4 +1,3 @@
-export * from './event-emitter.ts';
 export * from './exponential-backoff.ts';
 export * from './extended-json.ts';
 export * from './misc.ts';

source/sse-session/async-push-iterator.ts

@@ -23,24 +23,31 @@
  */
 export class AsyncPushIterator<T> {
     /** ReadableStream backing the async iterator returned from {@link Symbol.asyncIterator}. */
-    private readonly stream: ReadableStream<T>;
+    #stream: ReadableStream<T>;
 
     /** Controller used to enqueue values and close the stream from {@link push} and {@link close}. */
-    private controller: ReadableStreamDefaultController<T> | undefined;
+    #controller: ReadableStreamDefaultController<T> | undefined;
 
     /** When true, no more values are accepted and iteration eventually completes. */
-    public closed = false;
+    #closed = false;
 
     public constructor() {
         // `start`'s `this` is the underlying source object when using a plain method.
         // An arrow function captures the class instance so the controller is stored here.
-        this.stream = new ReadableStream({
+        this.#stream = new ReadableStream({
             start: (controller: ReadableStreamDefaultController<T>): void => {
-                this.controller = controller;
+                this.#controller = controller;
             },
         });
     }
 
+    /**
+     * Flag indicating if the iterator is closed.
+     */
+    public get closed(): boolean {
+        return this.#closed;
+    }
+
     /**
      * Enqueues a value for the consumer.
      *
@@ -49,9 +56,22 @@ export class AsyncPushIterator<T> {
      * @param value - The next value to yield from the iterator.
      */
     push(value: T): void {
-        if (this.closed) return;
+        if (this.#closed) return;
 
-        this.controller?.enqueue(value);
+        this.#controller?.enqueue(value);
+    }
+
+    /**
+     * Causes any future interactions with the associated stream to error with {@link error}.
+     * Calling this will also clear the pending values immediately, so iterators that were listening will not receive them.
+     *
+     * @param error - The error to throw from the stream.
+     */
+    error(error: Error): void {
+        if (this.#closed) return;
+
+        this.#closed = true;
+        this.#controller?.error(error);
     }
 
     /**
@@ -61,10 +81,10 @@ export class AsyncPushIterator<T> {
      * Buffered values are still yielded before iteration completes.
      */
     close(): void {
-        this.closed = true;
+        this.#closed = true;
 
         try {
-            this.controller?.close();
+            this.#controller?.close();
         } catch {
             // The reader may already have released or cancelled the stream.
         }
@@ -75,8 +95,11 @@ export class AsyncPushIterator<T> {
      *
      * Uses `preventCancel: true` so early `break` from `for await...of` does not
      * close the stream and block later pushes.
+     *
+     * Because values are discarded after being read, only a single consumer is supported.
+     * Additional consumers will receive a stream lock error. Unread values will be preserved until {@link close} is called.
      */
     [Symbol.asyncIterator](): AsyncIterableIterator<T> {
-        return this.stream.values({ preventCancel: true });
+        return this.#stream.values({ preventCancel: true });
     }
 }

source/sse-session/index.ts

@@ -1,4 +1,4 @@
+export * from './async-push-iterator.ts';
 export * from './types.ts';
 export * from './sse-session.ts';
 export * from './sse-event-parser.ts';
-export * from './async-push-iterator.ts';

source/sse-session/sse-session.ts

@@ -1,8 +1,9 @@
 import type { SSESessionOptions, SSESessionEventMap, SSEvent } from './types.ts';
 
+import { EventEmitter } from 'eventemitter3';
+
 import { HTTPError, ResponseBodyNullError } from '../errors.ts';
 import { tryAsync } from '../misc.ts';
-import { EventEmitter } from '../event-emitter.ts';
 import { ExponentialBackoff } from '../exponential-backoff.ts';
 
 import { SSEEventParser } from './sse-event-parser.ts';

test/event-emitter.test.ts

@@ -1,505 +0,0 @@
-import { expect, test, vi } from 'vitest';
-import { EventEmitter } from '../source/event-emitter.ts';
-
-/** Simple event map used across these tests. */
-type TestEvents = {
-    message: string;
-    count: number;
-};
-
-/**
- * Tests that EventEmitter invokes listeners when an event is emitted.
- */
-const testEventEmitterCallsListeners = (): void => {
-    const emitter = new EventEmitter<TestEvents>();
-    const listener = vi.fn();
-
-    // Register the listener and emit an event.
-    emitter.on('message', listener);
-    emitter.emit('message', 'hello');
-
-    // Expect the listener to have been called with the emitted payload.
-    expect(listener).toHaveBeenCalledOnce();
-    expect(listener).toHaveBeenCalledWith('hello');
-};
-
-/**
- * Tests that EventEmitter invokes all listeners registered for the same event.
- */
-const testEventEmitterCallsMultipleListeners = (): void => {
-    const emitter = new EventEmitter<TestEvents>();
-    const firstListener = vi.fn();
-    const secondListener = vi.fn();
-
-    // Register two listeners for the same event type.
-    emitter.on('count', firstListener);
-    emitter.on('count', secondListener);
-    emitter.emit('count', 42);
-
-    // Expect both listeners to receive the same payload.
-    expect(firstListener).toHaveBeenCalledOnce();
-    expect(firstListener).toHaveBeenCalledWith(42);
-    expect(secondListener).toHaveBeenCalledOnce();
-    expect(secondListener).toHaveBeenCalledWith(42);
-};
-
-/**
- * Tests that EventEmitter only invokes listeners registered for the emitted event type.
- */
-const testEventEmitterCallsOnlyMatchingListeners = (): void => {
-    const emitter = new EventEmitter<TestEvents>();
-    const messageListener = vi.fn();
-    const countListener = vi.fn();
-
-    // Register listeners on different event types.
-    emitter.on('message', messageListener);
-    emitter.on('count', countListener);
-
-    // Emit only the message event.
-    emitter.emit('message', 'hello');
-
-    // Expect only the matching listener to have been called.
-    expect(messageListener).toHaveBeenCalledOnce();
-    expect(countListener).not.toHaveBeenCalled();
-};
-
-/**
- * Tests that EventEmitter.emit returns false when no listeners are registered.
- */
-const testEventEmitterEmitReturnsFalseWithNoListeners = (): void => {
-    const emitter = new EventEmitter<TestEvents>();
-
-    const hasListeners = emitter.emit('message', 'hello');
-
-    // Expect emit to report that nobody was listening.
-    expect(hasListeners).toBe(false);
-};
-
-/**
- * Tests that EventEmitter.emit returns true when listeners are registered.
- */
-const testEventEmitterEmitReturnsTrueWithListeners = (): void => {
-    const emitter = new EventEmitter<TestEvents>();
-
-    emitter.on('message', vi.fn());
-
-    const hasListeners = emitter.emit('message', 'hello');
-
-    // Expect emit to report that at least one listener was invoked.
-    expect(hasListeners).toBe(true);
-};
-
-/**
- * Tests that the off callback returned by on() removes the listener.
- */
-const testEventEmitterOffCallbackRemovesListener = (): void => {
-    const emitter = new EventEmitter<TestEvents>();
-    const listener = vi.fn();
-
-    // on() returns an off callback that removes the listener.
-    const off = emitter.on('message', listener);
-    emitter.emit('message', 'first');
-
-    // Unsubscribe before emitting again.
-    off();
-    emitter.emit('message', 'second');
-
-    // Expect the listener to have only received the first event.
-    expect(listener).toHaveBeenCalledOnce();
-    expect(listener).toHaveBeenCalledWith('first');
-};
-
-/**
- * Tests that off() removes a listener when given the same function reference.
- */
-const testEventEmitterOffRemovesListenerByReference = (): void => {
-    const emitter = new EventEmitter<TestEvents>();
-    const listener = vi.fn();
-
-    emitter.on('message', listener);
-    emitter.off('message', listener);
-    emitter.emit('message', 'hello');
-
-    // Expect the listener to have been removed before the emit.
-    expect(listener).not.toHaveBeenCalled();
-};
-
-/**
- * Tests that off() does nothing when given an unknown listener reference.
- */
-const testEventEmitterOffIgnoresUnknownListener = (): void => {
-    const emitter = new EventEmitter<TestEvents>();
-    const listener = vi.fn();
-
-    emitter.on('message', listener);
-
-    // Try to remove a different function reference.
-    emitter.off('message', vi.fn());
-    emitter.emit('message', 'hello');
-
-    // Expect the original listener to still receive the event.
-    expect(listener).toHaveBeenCalledOnce();
-};
-
-/**
- * Tests that off() does nothing when called for an event type with no listeners.
- */
-const testEventEmitterOffIgnoresUnregisteredEventType = (): void => {
-    const emitter = new EventEmitter<TestEvents>();
-    const listener = vi.fn();
-
-    // Call off without ever registering this listener.
-    emitter.off('message', listener);
-
-    expect(listener).not.toHaveBeenCalled();
-};
-
-/**
- * Tests that once() listeners are invoked only one time.
- */
-const testEventEmitterOnceListenerFiresOnce = (): void => {
-    const emitter = new EventEmitter<TestEvents>();
-    const listener = vi.fn();
-
-    emitter.once('message', listener);
-    emitter.emit('message', 'first');
-    emitter.emit('message', 'second');
-
-    // Expect the listener to auto-unsubscribe after the first emit.
-    expect(listener).toHaveBeenCalledOnce();
-    expect(listener).toHaveBeenCalledWith('first');
-};
-
-/**
- * Tests that once() can be added when regular listeners already exist for the event type.
- */
-const testEventEmitterOnceWorksWithExistingListeners = (): void => {
-    const emitter = new EventEmitter<TestEvents>();
-    const existingListener = vi.fn();
-    const onceListener = vi.fn();
-
-    // Register a regular listener first so the event type already exists in the map.
-    emitter.on('message', existingListener);
-    emitter.once('message', onceListener);
-    emitter.emit('message', 'hello');
-
-    expect(existingListener).toHaveBeenCalledOnce();
-    expect(onceListener).toHaveBeenCalledOnce();
-};
-
-/**
- * Tests that the off callback returned by once() removes the listener before it fires.
- */
-const testEventEmitterOnceOffCallbackRemovesListener = (): void => {
-    const emitter = new EventEmitter<TestEvents>();
-    const listener = vi.fn();
-
-    const off = emitter.once('message', listener);
-
-    // Unsubscribe before the event is ever emitted.
-    off();
-    emitter.emit('message', 'hello');
-
-    expect(listener).not.toHaveBeenCalled();
-};
-
-/**
- * Tests that removeAllListeners() clears every registered listener.
- */
-const testEventEmitterRemoveAllListeners = (): void => {
-    const emitter = new EventEmitter<TestEvents>();
-    const messageListener = vi.fn();
-    const countListener = vi.fn();
-
-    emitter.on('message', messageListener);
-    emitter.on('count', countListener);
-    emitter.removeAllListeners();
-
-    // Emit on both event types after clearing all listeners.
-    emitter.emit('message', 'hello');
-    emitter.emit('count', 1);
-
-    expect(messageListener).not.toHaveBeenCalled();
-    expect(countListener).not.toHaveBeenCalled();
-};
-
-/**
- * Tests that waitFor() resolves when a matching event is emitted.
- */
-const testEventEmitterWaitForResolvesOnMatch = async (): Promise<void> => {
-    const emitter = new EventEmitter<TestEvents>();
-
-    // Wait until an event matches the predicate.
-    const waitPromise = emitter.waitFor('count', (payload) => payload === 42);
-
-    // Emit a non-matching event first, then the matching one.
-    emitter.emit('count', 41);
-    emitter.emit('count', 42);
-
-    await expect(waitPromise).resolves.toBe(42);
-};
-
-/**
- * Tests that waitFor() ignores non-matching events while other listeners still receive them.
- */
-const testEventEmitterWaitForIgnoresNonMatchingEvents = async (): Promise<void> => {
-    const emitter = new EventEmitter<TestEvents>();
-    const listener = vi.fn();
-
-    const waitPromise = emitter.waitFor('message', (payload) => payload === 'done');
-
-    // A regular listener should still receive every emit while waitFor filters.
-    emitter.on('message', listener);
-    emitter.emit('message', 'pending');
-    emitter.emit('message', 'done');
-
-    await expect(waitPromise).resolves.toBe('done');
-    expect(listener).toHaveBeenCalledTimes(2);
-};
-
-/**
- * Tests that waitFor() rejects when the timeout expires.
- */
-const testEventEmitterWaitForRejectsOnTimeout = async (): Promise<void> => {
-    vi.useFakeTimers();
-
-    try {
-        const emitter = new EventEmitter<TestEvents>();
-
-        const waitPromise = emitter.waitFor('message', () => true, 100);
-
-        // Attach the rejection handler before advancing timers so the rejection is handled.
-        const assertion = expect(waitPromise).rejects.toThrow('Timeout waiting for event "message"');
-
-        await vi.advanceTimersByTimeAsync(100);
-
-        await assertion;
-    } finally {
-        vi.useRealTimers();
-        vi.restoreAllMocks();
-    }
-};
-
-/**
- * Tests that waitFor() clears its timeout when it resolves before expiry.
- */
-const testEventEmitterWaitForClearsTimeoutOnResolve = async (): Promise<void> => {
-    vi.useFakeTimers();
-
-    try {
-        const emitter = new EventEmitter<TestEvents>();
-
-        // Register waitFor with a timeout, then resolve it before the timer fires.
-        const waitPromise = emitter.waitFor('message', (payload) => payload === 'done', 100);
-
-        emitter.emit('message', 'done');
-
-        await expect(waitPromise).resolves.toBe('done');
-
-        // If clearTimeout was not called, advancing past the timeout would reject the promise.
-        await vi.advanceTimersByTimeAsync(100);
-    } finally {
-        vi.useRealTimers();
-        vi.restoreAllMocks();
-    }
-};
-
-/**
- * Tests that waitFor() removes its listener after resolving.
- */
-const testEventEmitterWaitForRemovesListenerAfterResolve = async (): Promise<void> => {
-    const emitter = new EventEmitter<TestEvents>();
-
-    const waitPromise = emitter.waitFor('message', () => true);
-
-    emitter.emit('message', 'first');
-    await expect(waitPromise).resolves.toBe('first');
-
-    // Register a second waitFor so we can verify the first listener was cleaned up.
-    const secondWaitPromise = emitter.waitFor('message', (payload) => payload === 'second');
-
-    // Emit a payload that only the second waitFor should accept.
-    emitter.emit('message', 'ignored');
-
-    // Track whether the second waitFor resolves too early.
-    let resolvedEarly = false;
-    /* eslint-disable-next-line */
-    secondWaitPromise.then(() => {
-        resolvedEarly = true;
-    });
-
-    // Yield so any premature resolution would have a chance to run.
-    await Promise.resolve();
-    expect(resolvedEarly).toBe(false);
-
-    emitter.emit('message', 'second');
-    await expect(secondWaitPromise).resolves.toBe('second');
-};
-
-/**
- * Tests that the first debounced emit does not call clearTimeout.
- */
-const testEventEmitterDebouncedFirstEmitDoesNotClearTimeout = (): void => {
-    vi.useFakeTimers();
-
-    try {
-        const emitter = new EventEmitter<TestEvents>();
-        const clearTimeoutSpy = vi.spyOn(globalThis, 'clearTimeout');
-        const listener = vi.fn();
-
-        emitter.on('message', listener, 100);
-        emitter.emit('message', 'first');
-
-        // The first emit starts the debounce timer; there is nothing to clear yet.
-        expect(clearTimeoutSpy).not.toHaveBeenCalled();
-        expect(listener).not.toHaveBeenCalled();
-    } finally {
-        vi.useRealTimers();
-        vi.restoreAllMocks();
-    }
-};
-
-/**
- * Tests that debounced on() listeners receive only the last payload after the debounce window.
- */
-const testEventEmitterDebouncedOnListener = async (): Promise<void> => {
-    vi.useFakeTimers();
-
-    try {
-        const emitter = new EventEmitter<TestEvents>();
-        const listener = vi.fn();
-
-        emitter.on('message', listener, 100);
-
-        // Emit several events in quick succession.
-        emitter.emit('message', 'first');
-        emitter.emit('message', 'second');
-        emitter.emit('message', 'third');
-
-        // Expect the listener to not have fired yet.
-        expect(listener).not.toHaveBeenCalled();
-
-        // Advance past the debounce window.
-        await vi.advanceTimersByTimeAsync(100);
-
-        // Expect only the last payload to have been delivered.
-        expect(listener).toHaveBeenCalledOnce();
-        expect(listener).toHaveBeenCalledWith('third');
-    } finally {
-        vi.useRealTimers();
-        vi.restoreAllMocks();
-    }
-};
-
-/**
- * Tests that repeated debounced emits reset the debounce timer.
- */
-const testEventEmitterDebouncedTimerResetsOnRepeatedEmits = async (): Promise<void> => {
-    vi.useFakeTimers();
-
-    try {
-        const emitter = new EventEmitter<TestEvents>();
-        const listener = vi.fn();
-
-        emitter.on('count', listener, 100);
-        emitter.emit('count', 1);
-
-        // Advance halfway through the debounce window and emit again.
-        await vi.advanceTimersByTimeAsync(50);
-        emitter.emit('count', 2);
-        await vi.advanceTimersByTimeAsync(50);
-
-        // The timer was reset, so the listener should not have fired yet.
-        expect(listener).not.toHaveBeenCalled();
-
-        // Advance the remaining time for the reset timer to expire.
-        await vi.advanceTimersByTimeAsync(50);
-
-        expect(listener).toHaveBeenCalledOnce();
-        expect(listener).toHaveBeenCalledWith(2);
-    } finally {
-        vi.useRealTimers();
-        vi.restoreAllMocks();
-    }
-};
-
-/**
- * Tests that a debounce time of zero behaves like a normal listener.
- */
-const testEventEmitterZeroDebounceDoesNotDebounce = async (): Promise<void> => {
-    vi.useFakeTimers();
-
-    try {
-        const emitter = new EventEmitter<TestEvents>();
-        const listener = vi.fn();
-
-        // A debounce time of zero should behave like a normal listener.
-        emitter.on('message', listener, 0);
-        emitter.emit('message', 'first');
-        emitter.emit('message', 'second');
-
-        expect(listener).toHaveBeenCalledTimes(2);
-    } finally {
-        vi.useRealTimers();
-        vi.restoreAllMocks();
-    }
-};
-
-/**
- * Tests that debounced once() listeners fire once with the last payload.
- */
-const testEventEmitterDebouncedOnceListener = async (): Promise<void> => {
-    vi.useFakeTimers();
-
-    try {
-        const emitter = new EventEmitter<TestEvents>();
-        const listener = vi.fn();
-
-        emitter.once('message', listener, 100);
-        emitter.emit('message', 'first');
-        emitter.emit('message', 'second');
-
-        await vi.advanceTimersByTimeAsync(100);
-
-        // Expect the debounced once listener to fire once with the last payload.
-        expect(listener).toHaveBeenCalledOnce();
-        expect(listener).toHaveBeenCalledWith('second');
-
-        // Emit again after the debounce window; the once listener should stay removed.
-        emitter.emit('message', 'third');
-        await vi.advanceTimersByTimeAsync(100);
-
-        expect(listener).toHaveBeenCalledOnce();
-    } finally {
-        vi.useRealTimers();
-        vi.restoreAllMocks();
-    }
-};
-
-const runTests = async (): Promise<void> => {
-    test('EventEmitter: calls listeners when an event is emitted', testEventEmitterCallsListeners);
-    test('EventEmitter: calls multiple listeners for the same event', testEventEmitterCallsMultipleListeners);
-    test('EventEmitter: only calls listeners for the emitted event type', testEventEmitterCallsOnlyMatchingListeners);
-    test('EventEmitter: returns false when emitting with no listeners', testEventEmitterEmitReturnsFalseWithNoListeners);
-    test('EventEmitter: returns true when emitting with listeners', testEventEmitterEmitReturnsTrueWithListeners);
-    test('EventEmitter: stops calling a listener after its off callback is invoked', testEventEmitterOffCallbackRemovesListener);
-    test('EventEmitter: removes a listener when off is called with the same reference', testEventEmitterOffRemovesListenerByReference);
-    test('EventEmitter: ignores off when the listener reference is unknown', testEventEmitterOffIgnoresUnknownListener);
-    test('EventEmitter: ignores off for an event type with no listeners', testEventEmitterOffIgnoresUnregisteredEventType);
-    test('EventEmitter: calls a once listener only one time', testEventEmitterOnceListenerFiresOnce);
-    test('EventEmitter: registers once when listeners already exist', testEventEmitterOnceWorksWithExistingListeners);
-    test('EventEmitter: stops a once listener after its off callback is invoked', testEventEmitterOnceOffCallbackRemovesListener);
-    test('EventEmitter: removes all listeners when removeAllListeners is called', testEventEmitterRemoveAllListeners);
-    test('EventEmitter: resolves waitFor when a matching event is emitted', testEventEmitterWaitForResolvesOnMatch);
-    test('EventEmitter: ignores non-matching events while waiting with waitFor', testEventEmitterWaitForIgnoresNonMatchingEvents);
-    test('EventEmitter: rejects waitFor when the timeout is reached', testEventEmitterWaitForRejectsOnTimeout);
-    test('EventEmitter: clears the timeout when waitFor resolves before expiry', testEventEmitterWaitForClearsTimeoutOnResolve);
-    test('EventEmitter: removes the waitFor listener after it resolves', testEventEmitterWaitForRemovesListenerAfterResolve);
-    test('EventEmitter: does not clear a timeout on the first debounced emit', testEventEmitterDebouncedFirstEmitDoesNotClearTimeout);
-    test('EventEmitter: debounces on listeners', testEventEmitterDebouncedOnListener);
-    test('EventEmitter: resets the debounce timer on repeated emits', testEventEmitterDebouncedTimerResetsOnRepeatedEmits);
-    test('EventEmitter: does not debounce when debounceMilliseconds is zero', testEventEmitterZeroDebounceDoesNotDebounce);
-    test('EventEmitter: debounces once listeners and invokes them only once', testEventEmitterDebouncedOnceListener);
-};
-
-await runTests();

test/extended-json.test.ts

@@ -1,5 +1,5 @@
 import { expect, test } from 'vitest';
-import { extendedJsonReviver } from '../source/index.ts';
+import { extendedJsonReviver, toExtendedJson, fromExtendedJson } from '../source/index.ts';
 
 /**
  * Tests that extendedJsonReviver reconstructs a positive bigint.
@@ -64,6 +64,34 @@ const testReviverPassesThroughPlainString = (): void => {
     expect(reconstructed).toBe('just a string');
 };
 
+/**
+ * Tests that toExtendedJson serializes an object to a string.
+ */
+const testToExtendedJsonSerializesObject = (): void => {
+    // Define an object with a Uint8Array
+    const extendedObject = { bytes: new Uint8Array([ 0xab, 0xcd ]) };
+
+    // Serialize the object to a string
+    const serialized = toExtendedJson(extendedObject);
+
+    // The serialized string should contain the Uint8Array in Extended JSON format (Uint8Arrays are encoded as hex strings)
+    expect(serialized).toBe('{"bytes":"<uint8array: abcd>"}');
+};
+
+/**
+ * Tests that fromExtendedJson deserializes a string to an object.
+ */
+const testFromExtendedJsonDeserializesString = (): void => {
+    // Define a string in Extended JSON format that contains a Uint8Array
+    const serializedObject = '{"bytes":"<uint8array: abcd>"}';
+
+    // Deserialize the string to an object
+    const deserialized = fromExtendedJson(serializedObject);
+
+    // The deserialized object should contain the Uint8Array
+    expect(deserialized).toStrictEqual({ bytes: new Uint8Array([ 0xab, 0xcd ]) });
+};
+
 /**
  * Tests that extendedJsonReviver passes through non-string values.
  */
@@ -85,6 +113,8 @@ const runTests = async (): Promise<void> => {
     test('extendedJsonReviver: reconstructs an empty Uint8Array', testReviverReconstructsEmptyUint8Array);
     test('extendedJsonReviver: passes through plain strings', testReviverPassesThroughPlainString);
     test('extendedJsonReviver: passes through non-string values', testReviverPassesThroughNonStringValues);
+    test('toExtendedJson: serializes an object to a string', testToExtendedJsonSerializesObject);
+    test('fromExtendedJson: deserializes a string to an object', testFromExtendedJsonDeserializesString);
 };
 
 await runTests();

test/sse-session/async-push-iterator.test.ts

@@ -23,18 +23,14 @@ const collectAll = async <T>(iterator: AsyncPushIterator<T>): Promise<T[]> => {
 const testPushComposedPushAndConsume = async (): Promise<void> => {
     const iterator = new AsyncPushIterator<number>();
 
-    const result = new Promise<number[]>((resolve) => {
+    const result = (): Promise<number[]> => collectAll(iterator);
-        void (async (): Promise<void> => {
-            resolve(await collectAll(iterator));
-        })();
-    });
 
     iterator.push(1);
     iterator.push(2);
     iterator.push(3);
     iterator.close();
 
-    await expect(result).resolves.toEqual([ 1, 2, 3 ]);
+    await expect(result()).resolves.toEqual([ 1, 2, 3 ]);
 };
 
 /**
@@ -61,15 +57,11 @@ const testPushComposedBuffersValuesPushedBeforeLoopStarts = async (): Promise<vo
 const testPushComposedResolvesWithNoValues = async (): Promise<void> => {
     const iterator = new AsyncPushIterator<number>();
 
-    const result = new Promise<number[]>((resolve) => {
+    const result = async (): Promise<number[]> => collectAll(iterator);
-        void (async (): Promise<void> => {
-            resolve(await collectAll(iterator));
-        })();
-    });
 
     iterator.close();
 
-    await expect(result).resolves.toEqual([]);
+    await expect(result()).resolves.toEqual([]);
 };
 
 /**
@@ -78,11 +70,7 @@ const testPushComposedResolvesWithNoValues = async (): Promise<void> => {
 const testPushComposedIgnoresValuesAfterClose = async (): Promise<void> => {
     const iterator = new AsyncPushIterator<number>();
 
-    const result = new Promise<number[]>((resolve) => {
+    const result = async (): Promise<number[]> => collectAll(iterator);
-        void (async (): Promise<void> => {
-            resolve(await collectAll(iterator));
-        })();
-    });
 
     iterator.push(1);
     iterator.push(2);
@@ -90,7 +78,7 @@ const testPushComposedIgnoresValuesAfterClose = async (): Promise<void> => {
     iterator.close();
     iterator.push(4);
 
-    await expect(result).resolves.toEqual([ 1, 2, 3 ]);
+    await expect(result()).resolves.toEqual([ 1, 2, 3 ]);
 };
 
 /**
@@ -104,30 +92,19 @@ const testPushComposedRejectsMultipleConsumers = async (): Promise<void> => {
 
     const failureFlag = vi.fn();
 
-    const successfulIterator = (): Promise<number[]> =>
+    const successfulIterator = (): Promise<number[]> => collectAll(iterator);
-        new Promise((resolve) => {
-            void (async (): Promise<void> => {
-                resolve(await collectAll(iterator));
-            })();
-        });
 
-    const failedIterator = (): Promise<void> =>
+    const failedIterator = async (): Promise<void> => {
-        new Promise((resolve, reject) => {
+        try {
-            void (async (): Promise<void> => {
+            /* eslint-disable-next-line */
-                try {
+            for await (const _value of iterator) {
-                    /* eslint-disable-next-line */
+            }
-                    for await (const _value of iterator) {
+        } catch (error) {
-                    }
+            failureFlag();
-                } catch (error) {
+        }
-                    failureFlag();
+    };
-                    reject(error);
-                }
 
-                resolve();
+    const promises = [ successfulIterator(), failedIterator() ];
-            })();
-        });
-
-    const promises = [ successfulIterator(), failedIterator().catch(() => {}) ];
 
     iterator.close();
 
@@ -175,14 +152,63 @@ const testPushComposedAllowsPushingAfterEarlyBreak = async (): Promise<void> =>
     expect(secondPass).toEqual([ 2, 3 ]);
 };
 
+/**
+ * Tests that the iterator rejects after {@link AsyncPushIterator.error} is called.
+ */
+const testPushIteratorRejectsAfterError = async (): Promise<void> => {
+    const iterator = new AsyncPushIterator<number>();
+
+    iterator.error(new Error('Stream has been closed for a test'));
+
+    await expect(collectAll(iterator)).rejects.toThrow('Stream has been closed for a test');
+};
+
+/**
+ * Tests that the iterator closes the stream when error() is called.
+ */
+const testPushIteratorClosesWhenErrorIsCalled = async (): Promise<void> => {
+    const iterator = new AsyncPushIterator<number>();
+
+    iterator.error(new Error('Stream has been closed for a test'));
+    expect(iterator.closed).toBe(true);
+};
+
+/**
+ * Tests that subsequent calls to error() are ignored.
+ */
+const testPushIteratorIgnoresSubsequentErrorCalls = async (): Promise<void> => {
+    const iterator = new AsyncPushIterator<number>();
+
+    iterator.error(new Error('Stream has been closed for a test'));
+    expect(iterator.closed).toBe(true);
+
+    expect(iterator.error(new Error('Second error'))).toBe(undefined);
+};
+
+/**
+ * Tests that a consumer can check if the iterator is closed.
+ */
+const testPushIteratorCanCheckIfClosed = async (): Promise<void> => {
+    const iterator = new AsyncPushIterator<number>();
+
+    expect(iterator.closed).toBe(false);
+
+    iterator.close();
+    expect(iterator.closed).toBe(true);
+};
+
 const runTests = async (): Promise<void> => {
-    test('AsyncPushIterator (composed): pushes and consumes values', testPushComposedPushAndConsume);
+    test('AsyncPushIterator: pushes and consumes values', testPushComposedPushAndConsume);
-    test('AsyncPushIterator (composed): buffers values pushed before the for-await loop starts', testPushComposedBuffersValuesPushedBeforeLoopStarts);
+    test('AsyncPushIterator: buffers values pushed before the for-await loop starts', testPushComposedBuffersValuesPushedBeforeLoopStarts);
-    test('AsyncPushIterator (composed): resolves with no values when nothing was pushed', testPushComposedResolvesWithNoValues);
+    test('AsyncPushIterator: resolves with no values when nothing was pushed', testPushComposedResolvesWithNoValues);
-    test('AsyncPushIterator (composed): ignores values pushed after close', testPushComposedIgnoresValuesAfterClose);
+    test('AsyncPushIterator: ignores values pushed after close', testPushComposedIgnoresValuesAfterClose);
-    test('AsyncPushIterator (composed): rejects multiple consumers', testPushComposedRejectsMultipleConsumers);
+    test('AsyncPushIterator: rejects multiple consumers', testPushComposedRejectsMultipleConsumers);
-    test('AsyncPushIterator (composed): resolves immediately when closed before the loop starts', testPushComposedResolvesWhenClosedBeforeLoop);
+    test('AsyncPushIterator: resolves immediately when closed before the loop starts', testPushComposedResolvesWhenClosedBeforeLoop);
-    test('AsyncPushIterator (composed): keeps the stream open after an early break', testPushComposedAllowsPushingAfterEarlyBreak);
+    test('AsyncPushIterator: keeps the stream open after an early break', testPushComposedAllowsPushingAfterEarlyBreak);
+    test('AsyncPushIterator: rejects after error', testPushIteratorRejectsAfterError);
+    test('AsyncPushIterator: closes the stream when error() is called', testPushIteratorClosesWhenErrorIsCalled);
+    test('AsyncPushIterator: ignores subsequent error() calls', testPushIteratorIgnoresSubsequentErrorCalls);
+    test('AsyncPushIterator: can check if closed', testPushIteratorCanCheckIfClosed);
 };
 
 await runTests();