fix: propagate duplex transform sink errors (#8)

Sink errors were either getting smothered or causing unhandled promise rejections depending on your platform.

This breaks on Chrome when creating an abortable-iterator from a duplex transform stream that you then try to abort.

Instead if the sink of a duplex stream being used as a transform returns a promise, propagate errors thrown by the sink into it's source so they can be caught and handled by the consumer.

Author: achingbrain

package.json

@@ -120,11 +120,14 @@
   "license": "MIT",
   "devDependencies": {
     "aegir": "^36.1.3",
+    "delay": "^5.0.0",
     "it-all": "^1.0.6",
     "it-drain": "^1.0.5",
     "streaming-iterables": "^6.0.0"
   },
   "dependencies": {
+    "it-merge": "^1.0.4",
+    "it-pushable": "^2.0.0",
     "it-stream-types": "^1.0.3"
   },
   "repository": {

src/index.ts

@@ -1,3 +1,5 @@
+import { pushable } from 'it-pushable'
+import merge from 'it-merge'
 import type * as it from 'it-stream-types'
 
 export const rawPipe = (...fns: any) => {
@@ -22,7 +24,19 @@ export const isDuplex = <TSource, TSink = TSource, RSink = Promise<void>> (obj:
 
 const duplexPipelineFn = <TSource> (duplex: any) => {
   return (source: any): it.Source<TSource> => {
-    duplex.sink(source) // TODO: error on sink side is unhandled rejection - this is the same as pull streams
+    const p = duplex.sink(source)
+
+    if (p.then != null) {
+      const stream = pushable<TSource>()
+      p.then(() => {
+        stream.end()
+      }, (err: Error) => {
+        stream.end(err)
+      })
+
+      return merge(stream, duplex.source)
+    }
+
     return duplex.source
   }
 }

test/index.spec.ts

@@ -3,7 +3,8 @@ import { pipe } from '../src/index.js'
 import all from 'it-all'
 import drain from 'it-drain'
 import { filter, collect, consume } from 'streaming-iterables'
-import type { Duplex } from 'it-stream-types'
+import delay from 'delay'
+import type { Duplex, Source } from 'it-stream-types'
 
 const oneTwoThree = () => [1, 2, 3]
 
@@ -116,4 +117,24 @@ describe('it-pipe', () => {
 
     expect(result).to.be.undefined()
   })
+
+  it('should propagate duplex transform sink errors', async () => {
+    const err = new Error('Aaargh')
+
+    await expect(
+      pipe(
+        oneTwoThree, {
+          source: (async function * () {
+            await delay(1000)
+            yield 5
+          }()),
+          sink: async (source: Source<number>) => {
+            await delay(20)
+            throw err
+          }
+        },
+        async (source) => await drain(source)
+      )
+    ).to.eventually.be.rejected.with.property('message', err.message)
+  })
 })