fix: emit known and unknown errors onto stream

.changeset/weak-worms-flash.md

@@ -0,0 +1,5 @@
+---
+"windpipe": patch
+---
+
+fix: emit known and unknown errors onto node stream

src/stream/consumption.ts

@@ -186,6 +186,7 @@ export class StreamConsumption<T, E> extends StreamBase {
             for await (const atom of this) {
                 // Determine whether non-ok values should be filtered out
                 if (options?.atoms !== true && !isOk(atom)) {
+                    s.emit("error", atom.value);
                     continue;
                 }
 

test/consumption.test.ts

@@ -137,6 +137,63 @@ describe.concurrent("stream consumption", () => {
             expect(errors).toHaveLength(1);
         });
 
+        test("propagate known errors into raw readable stream", async ({ expect }) => {
+            expect.assertions(1);
+
+            const stream = promisifyStream(
+                $.from([$.ok("a"), $.ok("b"), $.error("some error"), $.ok("c")]).toReadable("raw"),
+            );
+
+            expect(stream).rejects.toEqual("some error");
+        });
+
+        test("propagate known errors into object readable stream", async ({ expect }) => {
+            expect.assertions(1);
+
+            const stream = promisifyStream(
+                $.from([$.ok("a"), $.ok("b"), $.error("some error"), $.ok("c")]).toReadable(
+                    "object",
+                ),
+            );
+
+            expect(stream).rejects.toEqual("some error");
+        });
+
+        test("propagate multiple errors into readable stream", async ({ expect }) => {
+            expect.assertions(2);
+
+            const stream = $.from([
+                $.ok("a"),
+                $.error("an error"),
+                $.ok("b"),
+                $.error("some error"),
+                $.ok("c"),
+            ]).toReadable("object");
+
+            // Monitor the stream
+            const { data, errors } = await emptyStream(stream);
+
+            expect(errors).toEqual(["an error", "some error"]);
+            expect(data).toEqual(["a", "b", "c"]);
+        });
+
+        test("propagate known and unknown errors", async ({ expect }) => {
+            expect.assertions(2);
+
+            const stream = $.from([
+                $.ok("a"),
+                $.error("an error"),
+                $.ok("b"),
+                $.unknown("unknown error", []),
+                $.ok("c"),
+            ]).toReadable("object");
+
+            // Monitor the stream
+            const { data, errors } = await emptyStream(stream);
+
+            expect(errors).toEqual(["an error", "unknown error"]);
+            expect(data).toEqual(["a", "b", "c"]);
+        });
     });
 });