fix: add initial reducer value; fix its types

Author: Falinor

src/index.test.ts

@@ -317,7 +317,7 @@ describe('Stream Utils', () => {
     })
 
     afterEach(() => {
-      vi.restoreAllMocks()
+      vi.useRealTimers()
     })
 
     it('should emit values at specified intervals', async () => {
@@ -366,7 +366,7 @@ describe('Stream Utils', () => {
   describe('reduce', () => {
     it('should reduce stream to a single value', async () => {
       const stream = fromIterable([1, 2, 3, 4, 5]).pipeThrough(
-        reduce((a, b) => a + b)
+        reduce((a, b) => a + b, 0)
       )
       const [sum] = await toArray(stream)
       expect(sum).toBe(15)
@@ -377,22 +377,22 @@ describe('Stream Utils', () => {
         reduce(async (a, b) => {
           await delay(10)
           return a + b
-        })
+        }, 0)
       )
       const [sum] = await toArray(stream)
       expect(sum).toBe(6)
     })
 
     it('should handle empty streams', async () => {
       const stream = fromIterable<number>([]).pipeThrough(
-        reduce((a, b) => a + b)
+        reduce((a, b) => a + b, 0)
       )
       const result = await toArray(stream)
       expect(result).toStrictEqual([])
     })
 
     it('should handle single value streams', async () => {
-      const stream = fromIterable([42]).pipeThrough(reduce((a, b) => a + b))
+      const stream = fromIterable([42]).pipeThrough(reduce((a, b) => a + b, 0))
       const [result] = await toArray(stream)
       expect(result).toBe(42)
     })

src/index.ts

@@ -351,6 +351,7 @@ export function interval(period: number): ReadableStream<number> {
  *
  * @category Transformation
  * @param reducer - The reducer function to apply to each chunk
+ * @param initialValue - The initial value
  * @returns A TransformStream that emits the final reduced value when the stream closes
  * @example
  * ```ts
@@ -359,20 +360,17 @@ export function interval(period: number): ReadableStream<number> {
  * const [sum] = await toArray(stream); // Gets the single reduced value
  * ```
  */
-export function reduce<T>(
-  reducer: (accumulator: T, chunk: T) => SyncOrAsync<T>
-): TransformStream<T, T> {
-  let accumulator: T | undefined
+export function reduce<T, R>(
+  reducer: (accumulator: R, chunk: T) => SyncOrAsync<R>,
+  initialValue: R
+): TransformStream<T, R> {
+  let accumulator: R | undefined
 
-  return new TransformStream<T, T>({
-    async transform(chunk: T, controller: TransformStreamDefaultController<T>) {
-      if (accumulator === undefined) {
-        accumulator = chunk
-      } else {
-        accumulator = await reducer(accumulator, chunk)
-      }
+  return new TransformStream<T, R>({
+    async transform(chunk) {
+      accumulator = await reducer(accumulator ?? initialValue, chunk)
     },
-    flush(controller: TransformStreamDefaultController<T>) {
+    flush(controller) {
       if (accumulator !== undefined) {
         controller.enqueue(accumulator)
       }
@@ -400,7 +398,7 @@ export function scan<T, R>(
 ): TransformStream<T, R> {
   let accumulator = initialValue
   return new TransformStream<T, R>({
-    async transform(chunk: T, controller: TransformStreamDefaultController<R>) {
+    async transform(chunk, controller) {
       accumulator = await scanner(accumulator, chunk)
       controller.enqueue(accumulator)
     }