fix: support goodbye sources

package.json

@@ -123,6 +123,7 @@
     "delay": "^5.0.0",
     "it-all": "^1.0.6",
     "it-drain": "^1.0.5",
+    "p-defer": "^4.0.0",
     "streaming-iterables": "^6.0.0"
   },
   "dependencies": {

src/index.ts

@@ -22,7 +22,7 @@ export const isDuplex = <TSource, TSink = TSource, RSink = Promise<void>> (obj:
   return obj != null && typeof obj.sink === 'function' && isIterable(obj.source)
 }
 
-const duplexPipelineFn = <TSource> (duplex: any) => {
+const duplexPipelineFn = <TSource> (duplex: it.Duplex<TSource>) => {
   return (source: any): it.Source<TSource> => {
     const p = duplex.sink(source)
 
@@ -34,7 +34,12 @@ const duplexPipelineFn = <TSource> (duplex: any) => {
         stream.end(err)
       })
 
-      return merge(stream, duplex.source)
+      const sourceWrap = async function * () {
+        yield * duplex.source
+        stream.end()
+      }
+
+      return merge(stream, sourceWrap())
     }
 
     return duplex.source

test/index.spec.ts

@@ -4,6 +4,7 @@ import all from 'it-all'
 import drain from 'it-drain'
 import { filter, collect, consume } from 'streaming-iterables'
 import delay from 'delay'
+import defer from 'p-defer'
 import type { Duplex, Source } from 'it-stream-types'
 
 const oneTwoThree = () => [1, 2, 3]
@@ -137,4 +138,39 @@ describe('it-pipe', () => {
       )
     ).to.eventually.be.rejected.with.property('message', err.message)
   })
+
+  it('should support goodbye-style source', async () => {
+    const deferred = defer()
+    const end = 5
+    let otherEnded = false
+
+    await pipe(
+      async function * () {
+        for (let i = 0; i < end + 1; i++) {
+          yield i
+        }
+
+        while (!otherEnded) { // eslint-disable-line no-unmodified-loop-condition
+          await delay(10)
+        }
+      }, {
+        sink: async (source: Source<number>) => {
+          for await (const val of source) {
+            if (val === end) {
+              deferred.resolve()
+            }
+          }
+        },
+        source: (async function * () {
+          yield * [1, 2, 3]
+          yield end
+          await deferred.promise
+        }())
+      },
+      async (source) => {
+        await drain(source)
+        otherEnded = true
+      }
+    )
+  })
 })