streamx - npm Package Compare versions

Comparing version 2.7.2 to 2.8.0

.github/workflows/test.yml

+name: Build Status
+
+on:
+  push:
+    branches:
+      - master
+  pull_request:
+    branches:
+      - master
+jobs:
+  build:
+    strategy:
+      matrix:
+        node-version: [10.x, 12.x, 14.x]
+        os: [ubuntu-16.04, macos-latest, windows-latest]
+    runs-on: ${{ matrix.os }}
+    steps:
+    - uses: actions/checkout@v2
+    - name: Use Node.js ${{ matrix.node-version }}
+      uses: actions/setup-node@v1
+      with:
+        node-version: ${{ matrix.node-version }}
+    - run: npm install
+    - run: npm test

test/byte-length.js

+const tape = require('tape')
+const { Readable, Writable } = require('../')
+
+const defaultSizes = [
+  { name: 'buf512', item: Buffer.alloc(512), size: 512 },
+  { name: 'number', item: 1, size: 1024 },
+  { name: 'number-byteLength', item: 1, size: 512, byteLength: () => 512 },
+  { name: 'number-byteLengthReadable', item: 1, size: 256, byteLength: () => 512, byteLengthExtended: () => 256 },
+  { name: 'uint8-512', item: new Uint8Array(512), size: 512 },
+  { name: 'uint32-64', item: new Uint32Array(64), size: 256 }
+]
+
+for (const { name, item, size, byteLength, byteLengthExtended } of defaultSizes) {
+  tape(`readable ${name}`, function (t) {
+    const r = new Readable({ byteLength, byteLengthReadable: byteLengthExtended })
+    r.push(item)
+    t.same(r._readableState.buffered, size)
+    t.end()
+  })
+  tape(`writable ${name}`, function (t) {
+    const w = new Writable({ byteLength, byteLengthWritable: byteLengthExtended })
+    w.write(item)
+    t.same(w._writableState.buffered, size)
+    t.end()
+  })
+}
+tape('byteLength receives readable item', function (t) {
+  const obj = {}
+  const r = new Readable({
+    byteLength: data => {
+      t.equals(obj, data)
+      t.end()
+    }
+  })
+  r.push(obj)
+})
+tape('byteLength receives writable item', function (t) {
+  const obj = {}
+  const r = new Writable({
+    byteLength: data => {
+      t.equals(obj, data)
+      return 1
+    }
+  })
+  r.write(obj)
+  t.end()
+})

index.js

@@ -520,2 +520,5 @@ const { EventEmitter } = require('events')
       if (opts.predestroy) this._predestroy = opts.predestroy
+      if (opts.signal) {
+        opts.signal.addEventListener('abort', abort.bind(this))
+      }
     }
@@ -637,6 +640,7 @@ }
 
-  static _fromAsyncIterator (ite) {
+  static _fromAsyncIterator (ite, opts) {
     let destroy
 
     const rs = new Readable({
+      ...opts,
       read (cb) {
@@ -661,4 +665,4 @@ ite.next().then(push).then(cb.bind(null, null)).catch(cb)
 
-  static from (data) {
-    if (data[asyncIterator]) return this._fromAsyncIterator(data[asyncIterator]())
+  static from (data, opts) {
+    if (data[asyncIterator]) return this._fromAsyncIterator(data[asyncIterator](), opts)
     if (!Array.isArray(data)) data = data === undefined ? [] : [data]
@@ -668,2 +672,3 @@
     return new Readable({
+      ...opts,
       read (cb) {
@@ -881,4 +886,8 @@ this.push(i === data.length ? null : data[i++])
 
+function isTypedArray (data) {
+  return typeof data === 'object' && data !== null && typeof data.byteLength === 'number'
+}
+
 function defaultByteLength (data) {
-  return Buffer.isBuffer(data) ? data.length : 1024
+  return isTypedArray(data) ? data.byteLength : 1024
 }
@@ -888,2 +897,6 @@
 
+function abort () {
+  this.destroy(new Error('Stream aborted.'))
+}
+
 module.exports = {
@@ -890,0 +903,0 @@ isStream,

package.json

 {
   "name": "streamx",
-  "version": "2.7.2",
+  "version": "2.8.0",
   "description": "An iteration of the Node.js core streams with a series of improvements",
@@ -11,2 +11,3 @@ "main": "index.js",
   "devDependencies": {
+    "abort-controller": "^3.0.0",
     "end-of-stream": "^1.4.1",
@@ -21,3 +22,3 @@ "standard": "^14.3.1",
     "type": "git",
-    "url": "https://github.com/mafintosh/streamx.git"
+    "url": "https://github.com/streamxorg/streamx.git"
   },
@@ -27,5 +28,5 @@ "author": "Mathias Buus (@mafintosh)",
   "bugs": {
-    "url": "https://github.com/mafintosh/streamx/issues"
+    "url": "https://github.com/streamxorg/streamx/issues"
   },
-  "homepage": "https://github.com/mafintosh/streamx"
+  "homepage": "https://github.com/streamxorg/streamx"
 }

README.md

@@ -9,3 +9,3 @@ # streamx
 
-[![Build Status](https://travis-ci.org/mafintosh/streamx.svg?branch=master)](https://travis-ci.org/mafintosh/streamx)
+[![Build Status](https://github.com/streamxorg/streamx/workflows/Build%20Status/badge.svg)](https://github.com/streamxorg/streamx/actions?query=workflow%3A%22Build+Status%22)
 
@@ -76,2 +76,8 @@ ## Main improvements from Node.js core stream
 
+#### AbortSignal support
+
+To make it easier to integrate streams in a `async/await` flow, all streams support a `signal` option
+that accepts a [`AbortSignal`](https://developer.mozilla.org/en-US/docs/Web/API/AbortSignal) to as an
+alternative means to `.destroy` streams.
+
 ## Usage
@@ -108,3 +114,4 @@
   map: (data) => data, // optional function to map input data
-  byteLength: (data) => size // optional function that calculates the byte size of input data
+  byteLength: (data) => size, // optional function that calculates the byte size of input data
+  signal: abortController.signal // optional AbortSignal that triggers `.destroy` when on `abort`
 }
@@ -257,3 +264,4 @@ ```
   map: (data) => data, // optional function to map input data
-  byteLength: (data) => size // optional function that calculates the byte size of input data
+  byteLength: (data) => size, // optional function that calculates the byte size of input data
+  signal: abortController.signal // optional AbortSignal that triggers `.destroy` when on `abort`
 }
@@ -260,0 +268,0 @@ ```

test/async-iterator.js

 const tape = require('tape')
 const { Readable } = require('../')
+const { AbortController } = require('abort-controller')
 
@@ -149,1 +150,24 @@ tape('streams are async iterators', async function (t) {
 })
+
+tape('using abort controller', async function (t) {
+  function createInfinite (signal) {
+    let count = 0
+    const r = new Readable({ signal })
+    r.push(count)
+    const int = setInterval(() => r.push(count++), 5000)
+    r.once('close', () => clearInterval(int))
+    return r
+  }
+  const controller = new AbortController()
+  const inc = []
+  setTimeout(() => controller.abort(), 10)
+  try {
+    for await (const chunk of createInfinite(controller.signal)) {
+      inc.push(chunk)
+    }
+  } catch (err) {
+    t.same(err.message, 'Stream aborted.')
+  }
+  t.same(inc, [0])
+  t.end()
+})

test/readable.js

@@ -120,1 +120,17 @@ const tape = require('tape')
 })
+
+tape('from array with highWaterMark', function (t) {
+  const r = Readable.from([1, 2, 3], { highWaterMark: 1 })
+  t.same(r._readableState.highWaterMark, 1)
+  t.end()
+})
+
+tape('from async iterator with highWaterMark', function (t) {
+  async function * test () {
+    yield 1
+  }
+
+  const r = Readable.from(test(), { highWaterMark: 1 })
+  t.same(r._readableState.highWaterMark, 1)
+  t.end()
+})

.travis.yml

-language: node_js
-node_js:
-  - '10'
-  - '12'
-os:
-  - windows
-  - osx
-  - linux

New alerts

Filesystem access

Supply chain risk

Accesses the file system, and could potentially read sensitive data.

Found 1 instance in 1 package

Fixed alerts

Filesystem access

Supply chain risk

Accesses the file system, and could potentially read sensitive data.

Found 1 instance in 1 package

Improved metrics

Total package byte prevSize

59459

6.79%

Number of package files

15

7.14%

Lines of code

1562

6.26%

Number of lines in readme file

404

2.02%

Worsened metrics

Dev dependency count

4

33.33%

No dependency changes