feat(nodejs-lib): harden pipeline mapper typings

packages/nodejs-lib/src/stream/pipeline.test.ts

@@ -1,4 +1,5 @@
 import { Readable } from 'node:stream'
+import { END } from '@naturalcycles/js-lib/types'
 import { expect, test } from 'vitest'
 import { Pipeline } from './pipeline.js'
 import type { ReadableTyped } from './stream.model.js'
@@ -16,6 +17,33 @@ test('Pipeline', async () => {
   expect(r).toEqual(['p_1', 'p_2', 'p_3'])
 })
 
+test('forEach returning END aborts the source', async () => {
+  const seen: string[] = []
+
+  await Pipeline.from<string>(new HonestReadable(100, 'p')).forEach(
+    async item => {
+      seen.push(item)
+      if (seen.length >= 5) return END
+    },
+    { concurrency: 1 },
+  )
+
+  expect(seen.length).toBeGreaterThanOrEqual(5)
+  expect(seen.length).toBeLessThan(100)
+})
+
+test('forEachSync returning END aborts the source', async () => {
+  const seen: string[] = []
+
+  await Pipeline.from<string>(new HonestReadable(100, 'p')).forEachSync(item => {
+    seen.push(item)
+    if (seen.length >= 5) return END
+  })
+
+  expect(seen.length).toBeGreaterThanOrEqual(5)
+  expect(seen.length).toBeLessThan(100)
+})
+
 /**
  * Readable that Honestly respects backpressure.
  */

packages/nodejs-lib/src/stream/pipeline.ts

@@ -8,6 +8,7 @@ import { createAbortableSignal } from '@naturalcycles/js-lib'
 import { _passthroughPredicate } from '@naturalcycles/js-lib/types'
 import type {
   AbortableAsyncMapper,
+  AbortableMapper,
   AsyncIndexedMapper,
   AsyncPredicate,
   END,
@@ -236,12 +237,12 @@ export class Pipeline<T = unknown> {
     return this
   }
 
-  tap(fn: AsyncIndexedMapper<T, any>, opt?: TransformOptions): this {
+  tap(fn: AsyncIndexedMapper<T, void>, opt?: TransformOptions): this {
     this.transforms.push(transformTap(fn, opt))
     return this
   }
 
-  tapSync(fn: IndexedMapper<T, any>, opt?: TransformOptions): this {
+  tapSync(fn: IndexedMapper<T, void>, opt?: TransformOptions): this {
     this.transforms.push(transformTapSync(fn, opt))
     return this
   }
@@ -420,7 +421,7 @@ export class Pipeline<T = unknown> {
   }
 
   async forEach(
-    fn: AsyncIndexedMapper<T, void>,
+    fn: AbortableAsyncMapper<T, void>,
     opt: TransformMapOptions<T, void> & TransformLogProgressOptions<T> = {},
   ): Promise<void> {
     this.transforms.push(
@@ -437,7 +438,7 @@ export class Pipeline<T = unknown> {
   }
 
   async forEachSync(
-    fn: IndexedMapper<T, void>,
+    fn: AbortableMapper<T, void>,
     opt: TransformMapSyncOptions<T, void> & TransformLogProgressOptions<T> = {},
   ): Promise<void> {
     this.transforms.push(