feat: allow for error/exception atoms to be rejected from toArray (#37

)

.changeset/green-balloons-sniff.md

@@ -0,0 +1,5 @@
+---
+"windpipe": patch
+---
+
+feat: reject error atom option for `toArray`

src/stream/consumption.ts

@@ -57,19 +57,22 @@ export class StreamConsumption<T, E> extends StreamBase {
      * Iterate through each atom in the stream, and return them as a single array.
      *
      * @param options.atoms - Return every atom on the stream.
+     * @param options.reject - If an error or exception is encountered, reject the promise with it.
      *
      * @group Consumption
      */
-    async toArray(options?: { atoms: false }): Promise<T[]>;
+    async toArray(options?: { atoms?: false; reject?: boolean }): Promise<T[]>;
     async toArray(options?: { atoms: true }): Promise<Atom<T, E>[]>;
-    async toArray(options?: { atoms?: boolean }): Promise<(Atom<T, E> | T)[]> {
+    async toArray(options?: { atoms?: boolean; reject?: boolean }): Promise<(Atom<T, E> | T)[]> {
         const array: (Atom<T, E> | T)[] = [];
 
         for await (const atom of this) {
             if (options?.atoms) {
                 array.push(atom);
             } else if (isOk(atom)) {
                 array.push(atom.value);
+            } else if (options?.reject) {
+                throw atom.value;
             }
         }
 

test/consumption.test.ts

@@ -71,6 +71,14 @@ describe.concurrent("stream consumption", () => {
 
             expect(array).toEqual([]);
         });
+
+        test("reject when error on stream", async ({ expect }) => {
+            expect.assertions(1);
+
+            const arrayPromise = $.from([$.ok(1), $.error("some error")]).toArray({ reject: true });
+
+            expect(arrayPromise).rejects.toThrow("some error");
+        });
     });
 
     describe.concurrent("toReadable", () => {