fix: continue emitting items after non-raw item encountered on raw st…

…ream

.changeset/heavy-insects-fly.md

@@ -0,0 +1,5 @@
+---
+"windpipe": patch
+---
+
+fix: continue emitting stream items after encountering non-raw item on raw stream

src/stream/consumption.ts

@@ -197,7 +197,7 @@ export class StreamConsumption<T, E> extends StreamBase {
                     const message = `Stream indicated it would emit raw values but emitted a '${typeof atom.value}' object`;
                     console.error(message);
                     s.emit("error", new Error(message));
-                    break;
+                    continue;
                 }
 
                 // Show a warning if any atom value is null

test/consumption.test.ts

@@ -125,6 +125,18 @@ describe.concurrent("stream consumption", () => {
             const streamPromise = promisifyStream(stream);
             expect(streamPromise).rejects.toBeTruthy();
         });
+
+        test("continue emitting items after non-raw item in raw stream", async ({ expect }) => {
+            expect.assertions(2);
+
+            const stream = $.from([1, "valid"]).toReadable("raw");
+
+            const { data, errors } = await emptyStream(stream);
+
+            expect(data).toHaveLength(1);
+            expect(errors).toHaveLength(1);
+        });
+
     });
 });
 
@@ -137,3 +149,16 @@ function promisifyStream(stream: Readable): Promise<unknown[]> {
         stream.on("end", () => resolve(data));
     });
 }
+
+async function emptyStream(stream: Readable): Promise<{ data: unknown[]; errors: unknown[] }> {
+    const errors: unknown[] = [];
+    const data: unknown[] = [];
+
+    await new Promise<void>((resolve) => {
+        stream.on("data", (d) => data.push(d));
+        stream.on("error", (e: string) => errors.push(e));
+        stream.on("end", () => resolve());
+    });
+
+    return { data, errors };
+}