fix: incorrect slicing when making batch (#33)

.changeset/pink-lamps-sit.md

@@ -0,0 +1,5 @@
+---
+"windpipe": patch
+---
+
+fix: incorrect splicing when generating batch

src/stream/transforms.ts

@@ -506,7 +506,7 @@ export class StreamTransforms<T, E> extends StreamConsumption<T, E> {
                         );
 
                         for (const batch of ready) {
-                            const items = batch.splice(0, options?.n ?? -1);
+                            const items = batch.splice(0, options?.n ?? batch.length);
                             yield ok<T[], E>(items);
                             totalBatchSize -= items.length;
                         }

test/transforms.test.ts

@@ -429,5 +429,34 @@ describe("stream transforms", () => {
                 [2, 4],
             ]);
         });
+
+        test("timeout bucket", async ({ expect }) => {
+            const mapper = vi.fn();
+
+            $.from([1, 2, 3, 4, 5, 6])
+                .batch({ timeout: 100, byBucket: (n) => (n % 2 === 0 ? "even" : "odd") })
+                .map(mapper)
+                .exhaust();
+
+            await vi.advanceTimersByTimeAsync(100);
+            expect(mapper).toHaveBeenCalledTimes(2);
+            expect(mapper).toHaveBeenNthCalledWith(1, [1, 3, 5]);
+            expect(mapper).toHaveBeenNthCalledWith(2, [2, 4, 6]);
+        });
+
+        test("timeout bucket no items", async ({ expect }) => {
+            const mapper = vi.fn();
+
+            $.fromNext(() => new Promise<number>(() => {}))
+                .batch({ timeout: 100, byBucket: (n) => (n % 2 === 0 ? "even" : "odd") })
+                .map(mapper)
+                .exhaust();
+
+            await vi.advanceTimersByTimeAsync(100);
+            expect(mapper).toHaveBeenCalledTimes(0);
+
+            await vi.advanceTimersByTimeAsync(100);
+            expect(mapper).toHaveBeenCalledTimes(0);
+        });
     });
 });