streamx - npm Package Compare versions

Comparing version 2.5.0 to 2.6.0

index.js

@@ -211,2 +211,4 @@ const { EventEmitter } = require('events')
   pipe (pipeTo, cb) {
+    if (this.pipeTo !== null) throw new Error('Can only pipe to one destination')
+
     this.stream._duplexState |= READ_PIPE_DRAINED
@@ -622,2 +624,38 @@ this.pipeTo = pipeTo
 
+  static _fromAsyncIterator (ite) {
+    let destroy
+
+    const rs = new Readable({
+      read (cb) {
+        ite.next().then(push).then(cb.bind(null, null)).catch(cb)
+      },
+      predestroy () {
+        destroy = ite.return()
+      },
+      destroy (cb) {
+        destroy.then(cb.bind(null, null)).catch(cb)
+      }
+    })
+
+    return rs
+
+    function push (data) {
+      if (data.done) rs.push(null)
+      else rs.push(data.value)
+    }
+  }
+
+  static from (data) {
+    if (data[asyncIterator]) return this._fromAsyncIterator(data[asyncIterator]())
+    if (!Array.isArray(data)) data = data === undefined ? [] : [data]
+
+    let i = 0
+    return new Readable({
+      read (cb) {
+        this.push(i === data.length ? null : data[i++])
+        cb(null)
+      }
+    })
+  }
+
   static isBackpressured (rs) {
@@ -641,2 +679,5 @@ return (rs._duplexState & READ_BACKPRESSURE_STATUS) !== 0 || rs._readableState.buffered >= rs._readableState.highWaterMark
     return {
+      [asyncIterator] () {
+        return this
+      },
       next () {
@@ -649,2 +690,5 @@ return new Promise(function (resolve, reject) {
         })
+      },
+      return () {
+        stream.destroy()
       }
@@ -651,0 +695,0 @@ }

package.json

 {
   "name": "streamx",
-  "version": "2.5.0",
+  "version": "2.6.0",
   "description": "An iteration of the Node.js core streams with a series of improvements",
@@ -5,0 +5,0 @@ "main": "index.js",

README.md

@@ -213,2 +213,6 @@ # streamx
 
+### `stream = Readable.from(arrayOrBufferOrStringOrAsyncIterator)
+
+Static method to turn an array or buffer or string or AsyncIterator into a readable stream.
+
 ## Writable Stream
@@ -215,0 +219,0 @@

test/readable.js

@@ -84,1 +84,37 @@ const tape = require('tape')
 })
+
+tape('from array', function (t) {
+  const inc = []
+  const r = Readable.from([1, 2, 3])
+  r.on('data', data => inc.push(data))
+  r.on('end', function () {
+    t.same(inc, [1, 2, 3])
+    t.end()
+  })
+})
+
+tape('from buffer', function (t) {
+  const inc = []
+  const r = Readable.from(Buffer.from('hello'))
+  r.on('data', data => inc.push(data))
+  r.on('end', function () {
+    t.same(inc, [Buffer.from('hello')])
+    t.end()
+  })
+})
+
+tape('from async iterator', function (t) {
+  async function * test () {
+    yield 1
+    yield 2
+    yield 3
+  }
+
+  const inc = []
+  const r = Readable.from(test())
+  r.on('data', data => inc.push(data))
+  r.on('end', function () {
+    t.same(inc, [1, 2, 3])
+    t.end()
+  })
+})

New alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Fixed alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Improved metrics

Total package byte prevSize

50666

increased by3.76%

Lines of code

1293

increased by5.64%

Number of lines in readme file

367

increased by1.1%

No dependency changes