From 70fe27305ec03d6a5a09eef02a2a53ba0c32bc00 Mon Sep 17 00:00:00 2001
From: Tom Anderson <tom@ando.gq>
Date: Tue, 16 Jul 2024 16:48:32 +1000
Subject: [PATCH] feat: allow for error/exception atoms to be rejected from
 `toArray`

---
 .changeset/green-balloons-sniff.md | 5 +++++
 src/stream/consumption.ts          | 7 +++++--
 test/consumption.test.ts           | 8 ++++++++
 3 files changed, 18 insertions(+), 2 deletions(-)
 create mode 100644 .changeset/green-balloons-sniff.md

diff --git a/.changeset/green-balloons-sniff.md b/.changeset/green-balloons-sniff.md
new file mode 100644
index 0000000..8ad85a7
--- /dev/null
+++ b/.changeset/green-balloons-sniff.md
@@ -0,0 +1,5 @@
+---
+"windpipe": patch
+---
+
+feat: reject error atom option for `toArray`
diff --git a/src/stream/consumption.ts b/src/stream/consumption.ts
index dbb0aa2..14adc0f 100644
--- a/src/stream/consumption.ts
+++ b/src/stream/consumption.ts
@@ -57,12 +57,13 @@ 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) {
@@ -70,6 +71,8 @@ export class StreamConsumption<T, E> extends StreamBase {
                 array.push(atom);
             } else if (isOk(atom)) {
                 array.push(atom.value);
+            } else if (options?.reject) {
+                throw atom.value;
             }
         }
 
diff --git a/test/consumption.test.ts b/test/consumption.test.ts
index 6569f6e..30dcbb3 100644
--- a/test/consumption.test.ts
+++ b/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", () => {