Merge pull request #1 from clear/feat/drop

add `drop` stream transform

.changeset/spotty-otters-sniff.md

@@ -0,0 +1,5 @@
+---
+"windpipe": minor
+---
+
+add `drop` stream transform

src/stream/transforms.ts

@@ -226,6 +226,33 @@ export class StreamTransforms<T, E> extends StreamConsumption<T, E> {
         });
     }
 
+    /**
+     * Drop the first `n` items from the stream.
+     *
+     * @group Transform
+     */
+    drop(n: number, options?: { atoms?: boolean }): Stream<T, E> {
+        this.trace("drop");
+
+        return this.consume(async function* (it) {
+            let i = 0;
+
+            for await (const atom of it) {
+                // Skip this atom if only values are desired
+                if (!options?.atoms && !isOk(atom)) {
+                    continue;
+                }
+
+                // Only yield if we're beyond the first n items
+                if (i >= n) {
+                    yield atom;
+                }
+
+                i++;
+            }
+        });
+    }
+
     /**
      * Delay emitting each value on the stream by `ms`.
      *

test/index.test.ts

@@ -167,6 +167,40 @@ describe.concurrent("stream transforms", () => {
             expect(await s.toArray({ atoms: true })).toEqual([error("an error"), ok(2), ok(4)]);
         });
     });
+
+    describe.concurrent("drop", () => {
+        test("multiple values", async ({ expect }) => {
+            expect.assertions(1);
+
+            const s = Stream.from([1, 2, 3, 4, 5]).drop(2);
+
+            expect(await s.toArray({ atoms: true })).toEqual([ok(3), ok(4), ok(5)]);
+        });
+
+        test("multiple values with errors", async ({ expect }) => {
+            expect.assertions(1);
+
+            const s = Stream.from([1, error("some error"), 2, 3, 4, 5]).drop(2);
+
+            expect(await s.toArray({ atoms: true })).toEqual([ok(3), ok(4), ok(5)]);
+        });
+
+        test("multiple atoms", async ({ expect }) => {
+            expect.assertions(1);
+
+            const s = Stream.from([1, 2, 3, 4, 5]).drop(2, { atoms: true });
+
+            expect(await s.toArray({ atoms: true })).toEqual([ok(3), ok(4), ok(5)]);
+        });
+
+        test("multiple atoms with errors", async ({ expect }) => {
+            expect.assertions(1);
+
+            const s = Stream.from([1, error("some error"), 2, 3, 4, 5]).drop(2, { atoms: true });
+
+            expect(await s.toArray({ atoms: true })).toEqual([ok(2), ok(3), ok(4), ok(5)]);
+        });
+    });
 });
 
 describe.concurrent("error handling", () => {