async iterator support

README.md

@@ -75,6 +75,24 @@ function factory (cb) {
 new MultiStream(factory).pipe(process.stdout) // => 123
 ```
 
+Async iterator support:
+
+```js
+const delay = require('util').promisify(setTimeout)
+const fs = require('fs')
+
+async function * generate() {
+  yield fs.createReadStream(__dirname + '/numbers/1.txt')
+  yield fs.createReadStream(__dirname + '/numbers/2.txt')
+
+  await delay(1000)
+
+  yield fs.createReadStream(__dirname + '/numbers/3.txt') 
+}
+
+new MultiStream(generate()).pipe(process.stdout)
+```
+
 ### contributors
 
 - [Feross Aboukhadijeh](http://feross.org)

index.js

@@ -2,6 +2,10 @@
 const stream = require('readable-stream')
 const once = require('once')
 
+function isAsyncIterator (obj) {
+  return typeof obj[Symbol.asyncIterator] === 'function'
+}
+
 function toStreams2Obj (s) {
   return toStreams2(s, { objectMode: true, highWaterMark: 16 })
 }
@@ -29,13 +33,13 @@ class MultiStream extends stream.Readable {
     this._current = null
     this._toStreams2 = (opts && opts.objectMode) ? toStreams2Obj : toStreams2Buf
 
-    if (typeof streams === 'function') {
-      this._queue = streams
-    } else {
+    if (streams instanceof Array) {
       this._queue = streams.map(this._toStreams2)
       this._queue.forEach(stream => {
         if (typeof stream !== 'function') this._attachErrorListener(stream)
       })
+    } else {
+      this._queue = streams
     }
 
     this._next()
@@ -61,7 +65,7 @@ class MultiStream extends stream.Readable {
   _destroy (err, cb) {
     let streams = []
     if (this._current) streams.push(this._current)
-    if (typeof this._queue !== 'function') streams = streams.concat(this._queue)
+    if (this._queue instanceof Array) streams = streams.concat(this._queue)
 
     if (streams.length === 0) {
       cb(err)
@@ -82,20 +86,36 @@ class MultiStream extends stream.Readable {
   _next () {
     this._current = null
 
+    if (this._queue instanceof Array) {
+      let stream = this._queue.shift()
+      if (typeof stream === 'function') {
+        stream = this._toStreams2(stream())
+        this._attachErrorListener(stream)
+      }
+      this._gotNextStream(stream)
+      return
+    }
+
+    if (isAsyncIterator(this._queue)) {
+      this._queue.next().then(({ value, done }) => {
+        if (done) {
+          this.push(null)
+        } else {
+          const stream = this._toStreams2(value)
+          this._attachErrorListener(stream)
+          this._gotNextStream(stream)
+        }
+      }).catch(err => this.destroy(err))
+      return
+    }
+
     if (typeof this._queue === 'function') {
       this._queue((err, stream) => {
         if (err) return this.destroy(err)
         stream = this._toStreams2(stream)
         this._attachErrorListener(stream)
         this._gotNextStream(stream)
       })
-    } else {
-      let stream = this._queue.shift()
-      if (typeof stream === 'function') {
-        stream = this._toStreams2(stream())
-        this._attachErrorListener(stream)
-      }
-      this._gotNextStream(stream)
     }
   }
 

test/basic.js

@@ -151,3 +151,23 @@ test('throw immediate error', function (t) {
 
   streams[1].emit('error', new Error('immediate error!'))
 })
+
+test('async iterator', function (t) {
+
+  async function * generate () {
+    yield str('1')
+    yield str('2')
+    yield str('3')
+  }
+
+  const stream = new MultiStream(generate())
+    .on('error', function (err) {
+      t.fail(err)
+    })
+
+  concat(stream, function (err, data) {
+    t.error(err)
+    t.equal(data.toString(), '123')
+    t.end()
+  })
+})