Merge branch 'development' into sse-and-backoff

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) => {
-        void (async (): Promise<void> => {
-            resolve(await collectAll(iterator));
-        })();
-    });
+    const result = (): Promise<number[]> => 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) => {
-        void (async (): Promise<void> => {
-            resolve(await collectAll(iterator));
-        })();
-    });
+    const result = async (): Promise<number[]> => 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) => {
-        void (async (): Promise<void> => {
-            resolve(await collectAll(iterator));
-        })();
-    });
+    const result = async (): Promise<number[]> => 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[]> =>
-        new Promise((resolve) => {
-            void (async (): Promise<void> => {
-                resolve(await collectAll(iterator));
-            })();
-        });
+    const successfulIterator = (): Promise<number[]> => collectAll(iterator);
 
-    const failedIterator = (): Promise<void> =>
-        new Promise((resolve, reject) => {
-            void (async (): Promise<void> => {
-                try {
-                    /* eslint-disable-next-line */
-                    for await (const _value of iterator) {
-                    }
-                } catch (error) {
-                    failureFlag();
-                    reject(error);
-                }
+    const failedIterator = async (): Promise<void> => {
+        try {
+            /* eslint-disable-next-line */
+            for await (const _value of iterator) {
+            }
+        } catch (error) {
+            failureFlag();
+        }
+    };
 
-                resolve();
-            })();
-        });
-
-    const promises = [ successfulIterator(), failedIterator().catch(() => {}) ];
+    const promises = [ successfulIterator(), failedIterator() ];
 
     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 (composed): buffers values pushed before the for-await loop starts', testPushComposedBuffersValuesPushedBeforeLoopStarts);
-    test('AsyncPushIterator (composed): resolves with no values when nothing was pushed', testPushComposedResolvesWithNoValues);
-    test('AsyncPushIterator (composed): ignores values pushed after close', testPushComposedIgnoresValuesAfterClose);
-    test('AsyncPushIterator (composed): rejects multiple consumers', testPushComposedRejectsMultipleConsumers);
-    test('AsyncPushIterator (composed): resolves immediately when closed before the loop starts', testPushComposedResolvesWhenClosedBeforeLoop);
-    test('AsyncPushIterator (composed): keeps the stream open after an early break', testPushComposedAllowsPushingAfterEarlyBreak);
+    test('AsyncPushIterator: pushes and consumes values', testPushComposedPushAndConsume);
+    test('AsyncPushIterator: buffers values pushed before the for-await loop starts', testPushComposedBuffersValuesPushedBeforeLoopStarts);
+    test('AsyncPushIterator: resolves with no values when nothing was pushed', testPushComposedResolvesWithNoValues);
+    test('AsyncPushIterator: ignores values pushed after close', testPushComposedIgnoresValuesAfterClose);
+    test('AsyncPushIterator: rejects multiple consumers', testPushComposedRejectsMultipleConsumers);
+    test('AsyncPushIterator: resolves immediately when closed before the loop starts', testPushComposedResolvesWhenClosedBeforeLoop);
+    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();