support setEncoding on readables (#89)

index.js

@@ -4,6 +4,7 @@ const PREMATURE_CLOSE = new Error('Premature close')
 
 const queueTick = require('queue-tick')
 const FIFO = require('fast-fifo')
+const TextDecoder = require('text-decoder')
 
 /* eslint-disable no-multi-spaces */
 
@@ -289,7 +290,11 @@ class ReadableState {
       return false
     }
 
-    if (this.map !== null) data = this.map(data)
+    if (this.map !== null) {
+      data = this.map(data)
+      if (data === null) return this.buffered < this.highWaterMark
+    }
+
     this.buffered += this.byteLength(data)
     this.queue.push(data)
 
@@ -686,6 +691,19 @@ class Readable extends Stream {
       if (this._readableState.readAhead === false) this._duplexState &= READ_NO_READ_AHEAD
       if (opts.read) this._read = opts.read
       if (opts.eagerOpen) this._readableState.updateNextTick()
+      if (opts.encoding) this.setEncoding(opts.encoding)
+    }
+  }
+
+  setEncoding (encoding) {
+    const dec = new TextDecoder(encoding)
+    const map = this._readableState.map || echo
+    this._readableState.map = mapOrSkip
+    return this
+
+    function mapOrSkip (data) {
+      const next = dec.push(data)
+      return next === '' ? null : map(next)
     }
   }
 
@@ -1080,6 +1098,10 @@ function pipeline (stream, ...streams) {
   }
 }
 
+function echo (s) {
+  return s
+}
+
 function isStream (stream) {
   return !!stream._readableState || !!stream._writableState
 }

package.json

@@ -4,10 +4,12 @@
   "description": "An iteration of the Node.js core streams with a series of improvements",
   "main": "index.js",
   "dependencies": {
-    "fast-fifo": "^1.1.0",
-    "queue-tick": "^1.0.1"
+    "fast-fifo": "^1.3.2",
+    "queue-tick": "^1.0.1",
+    "text-decoder": "^1.1.0"
   },
   "devDependencies": {
+    "b4a": "^1.6.6",
     "brittle": "^3.1.1",
     "end-of-stream": "^1.4.4",
     "standard": "^17.0.0"

test/readable.js

@@ -1,4 +1,5 @@
 const test = require('brittle')
+const b4a = require('b4a')
 const { Readable } = require('../')
 
 test('ondata', function (t) {
@@ -355,6 +356,38 @@ test('no read-ahead with async iterator', async function (t) {
   t.is(expectedTick, 10)
 })
 
+test('setEncoding', async function (t) {
+  const r = new Readable()
+
+  r.setEncoding('utf-8')
+  const buffer = b4a.from('hællå wørld!')
+  for (let i = 0; i < buffer.byteLength; i++) {
+    r.push(buffer.subarray(i, i + 1))
+  }
+  r.push(null)
+  const expected = b4a.toString(buffer).split('')
+  for await (const data of r) {
+    t.is(data, expected.shift())
+  }
+  t.is(expected.length, 0)
+})
+
+test('setEncoding respects existing map', async function (t) {
+  t.plan(1)
+
+  const r = new Readable({
+    encoding: 'utf-8',
+    map (data) {
+      return JSON.parse(data)
+    }
+  })
+
+  r.push('{"hello":"world"}')
+  r.once('data', function (data) {
+    t.alike(data, { hello: 'world' })
+  })
+})
+
 function nextImmediate () {
   return new Promise(resolve => setImmediate(resolve))
 }