feature: add async behavior opt-in

Author: isaacs

README.md

@@ -63,6 +63,10 @@ some ways superior to) Node.js core streams.
 Please read these caveats if you are familiar with node-core streams and
 intend to use Minipass streams in your programs.
 
+You can avoid most of these differences entirely (for a very
+small performance penalty) by setting `{async: true}` in the
+constructor options.
+
 ### Timing
 
 Minipass streams are designed to support synchronous use-cases.  Thus, data
@@ -82,6 +86,82 @@ This non-deferring approach makes Minipass streams much easier to reason
 about, especially in the context of Promises and other flow-control
 mechanisms.
 
+Example:
+
+```js
+const Minipass = require('minipass')
+const stream = new Minipass({ async: true })
+stream.on('data', () => console.log('data event'))
+console.log('before write')
+stream.write('hello')
+console.log('after write')
+// output:
+// before write
+// data event
+// after write
+```
+
+### Exception: Async Opt-In
+
+If you wish to have a Minipass stream with behavior that more
+closely mimics Node.js core streams, you can set the stream in
+async mode either by setting `async: true` in the constructor
+options, or by setting `stream.async = true` later on.
+
+```js
+const Minipass = require('minipass')
+const asyncStream = new Minipass({ async: true })
+asyncStream.on('data', () => console.log('data event'))
+console.log('before write')
+asyncStream.write('hello')
+console.log('after write')
+// output:
+// before write
+// after write
+// data event <-- this is deferred until the next tick
+```
+
+Switching _out_ of async mode is unsafe, as it could cause data
+corruption, and so is not enabled.  Example:
+
+```js
+const Minipass = require('minipass')
+const stream = new Minipass({ encoding: 'utf8' })
+stream.on('data', chunk => console.log(chunk))
+stream.async = true
+console.log('before writes')
+stream.write('hello')
+setStreamSyncAgainSomehow(stream) // <-- this doesn't actually exist!
+stream.write('world')
+console.log('after writes')
+// hypothetical output would be:
+// before writes
+// world
+// after writes
+// hello
+// NOT GOOD!
+```
+
+To avoid this problem, once set into async mode, any attempt to
+make the stream sync again will be ignored.
+
+```js
+const Minipass = require('minipass')
+const stream = new Minipass({ encoding: 'utf8' })
+stream.on('data', chunk => console.log(chunk))
+stream.async = true
+console.log('before writes')
+stream.write('hello')
+stream.async = false // <-- no-op, stream already async
+stream.write('world')
+console.log('after writes')
+// actual output:
+// before writes
+// after writes
+// hello
+// world
+```
+
 ### No High/Low Water Marks
 
 Node.js core streams will optimistically fill up a buffer, returning `true`
@@ -97,6 +177,9 @@ If the data has nowhere to go, then `write()` returns false, and the data
 sits in a buffer, to be drained out immediately as soon as anyone consumes
 it.
 
+Since nothing is ever buffered unnecessarily, there is much less
+copying data, and less bookkeeping about buffer capacity levels.
+
 ### Hazards of Buffering (or: Why Minipass Is So Fast)
 
 Since data written to a Minipass stream is immediately written all the way
@@ -181,6 +264,8 @@ moving on to the next entry in an archive parse stream, etc.) then be sure
 to call `stream.pause()` on creation, and then `stream.resume()` once you
 are ready to respond to the `end` event.
 
+However, this is _usually_ not a problem because:
+
 ### Emit `end` When Asked
 
 One hazard of immediately emitting `'end'` is that you may not yet have had
@@ -197,6 +282,18 @@ To prevent calling handlers multiple times who would not expect multiple
 ends to occur, all listeners are removed from the `'end'` event whenever it
 is emitted.
 
+### Emit `error` When Asked
+
+The most recent error object passed to the `'error'` event is
+stored on the stream.  If a new `'error'` event handler is added,
+and an error was previously emitted, then the event handler will
+be called immediately (or on `process.nextTick` in the case of
+async streams).
+
+This makes it much more difficult to end up trying to interact
+with a broken stream, if the error handler is added after an
+error was previously emitted.
+
 ### Impact of "immediate flow" on Tee-streams
 
 A "tee stream" is a stream piping to multiple destinations:
@@ -221,7 +318,7 @@ src.pipe(dest1) // 'foo' chunk flows to dest1 immediately, and is gone
 src.pipe(dest2) // gets nothing!
 ```
 
-The solution is to create a dedicated tee-stream junction that pipes to
+One solution is to create a dedicated tee-stream junction that pipes to
 both locations, and then pipe to _that_ instead.
 
 ```js
@@ -258,6 +355,13 @@ tee.on('data', handler2)
 src.pipe(tee)
 ```
 
+All of the hazards in this section are avoided by setting `{
+async: true }` in the Minipass constructor, or by setting
+`stream.async = true` afterwards.  Note that this does add some
+overhead, so should only be done in cases where you are willing
+to lose a bit of performance in order to avoid having to refactor
+program logic.
+
 ## USAGE
 
 It's a stream!  Use it like a stream and it'll most likely do what you

index.js

@@ -28,6 +28,9 @@ const OBJECTMODE = Symbol('objectMode')
 const DESTROYED = Symbol('destroyed')
 const EMITDATA = Symbol('emitData')
 const EMITEND = Symbol('emitData')
+const ASYNC = Symbol('async')
+
+const defer = fn => process.nextTick(fn)
 
 // TODO remove when Node v8 support drops
 const doIter = global._MP_NO_ITERATOR_SYMBOLS_  !== '1'
@@ -82,6 +85,7 @@ module.exports = class Minipass extends Stream {
       this[ENCODING] = options && options.encoding || null
     if (this[ENCODING] === 'buffer')
       this[ENCODING] = null
+    this[ASYNC] = options && !!options.async || false
     this[DECODER] = this[ENCODING] ? new SD(this[ENCODING]) : null
     this[EOF] = false
     this[EMITTED_END] = false
@@ -121,6 +125,9 @@ module.exports = class Minipass extends Stream {
   get objectMode () { return this[OBJECTMODE] }
   set objectMode (om) { this[OBJECTMODE] = this[OBJECTMODE] || !!om }
 
+  get ['async'] () { return this[ASYNC] }
+  set ['async'] (a) { this[ASYNC] = this[ASYNC] || !!a }
+
   write (chunk, encoding, cb) {
     if (this[EOF])
       throw new Error('write after end')
@@ -139,6 +146,8 @@ module.exports = class Minipass extends Stream {
     if (!encoding)
       encoding = 'utf8'
 
+    const fn = this[ASYNC] ? defer : f => f()
+
     // convert array buffers and typed array views into buffers
     // at some point in the future, we may want to do the opposite!
     // leave strings and buffers as-is
@@ -169,7 +178,7 @@ module.exports = class Minipass extends Stream {
         this.emit('readable')
 
       if (cb)
-        cb()
+        fn(cb)
 
       return this.flowing
     }
@@ -180,7 +189,7 @@ module.exports = class Minipass extends Stream {
       if (this[BUFFERLENGTH] !== 0)
         this.emit('readable')
       if (cb)
-        cb()
+        fn(cb)
       return this.flowing
     }
 
@@ -208,7 +217,7 @@ module.exports = class Minipass extends Stream {
       this.emit('readable')
 
     if (cb)
-      cb()
+      fn(cb)
 
     return this.flowing
   }
@@ -358,7 +367,10 @@ module.exports = class Minipass extends Stream {
         dest.end()
     } else {
       this.pipes.push(new Pipe(this, dest, opts))
-      this[RESUME]()
+      if (this[ASYNC])
+        defer(() => this[RESUME]())
+      else
+        this[RESUME]()
     }
 
     return dest
@@ -378,7 +390,10 @@ module.exports = class Minipass extends Stream {
       super.emit(ev)
       this.removeAllListeners(ev)
     } else if (ev === 'error' && this[EMITTED_ERROR]) {
-      fn.call(this, this[EMITTED_ERROR])
+      if (this[ASYNC])
+        defer(() => fn.call(this, this[EMITTED_ERROR]))
+      else
+        fn.call(this, this[EMITTED_ERROR])
     }
     return ret
   }
@@ -408,10 +423,14 @@ module.exports = class Minipass extends Stream {
     if (ev !== 'error' && ev !== 'close' && ev !== DESTROYED && this[DESTROYED])
       return
     else if (ev === 'data') {
-      return data ? this[EMITDATA](data) : false
+      return !data ? false
+        : this[ASYNC] ? defer(() => this[EMITDATA](data))
+        : this[EMITDATA](data)
     } else if (ev === 'end') {
       // only actual end gets this treatment
-      return this[EMITTED_END] ? false : this[EMITEND]()
+      return this[EMITTED_END] ? false
+        : this[ASYNC] ? defer(() => this[EMITEND]())
+        : this[EMITEND]()
     } else if (ev === 'close') {
       this[CLOSED] = true
       // don't emit close before 'end' and 'finish'

test/async-stream.js

@@ -0,0 +1,89 @@
+const t = require('tap')
+const MP = require('../')
+
+t.test('pipe', t => {
+  const m = new MP({ encoding: 'utf8', async: true })
+  const d1 = new MP({ encoding: 'utf8' })
+  const d2 = new MP({ encoding: 'utf8' })
+
+  m.pipe(d1)
+  m.write('hello, ')
+  m.pipe(d2)
+  m.write('world')
+  m.end()
+  return Promise.all([d1.concat(), d2.concat()])
+    .then(result => t.strictSame(result, ['hello, world', 'hello, world']))
+})
+
+t.test('pipe split', t => {
+  const m = new MP({ encoding: 'utf8' })
+  t.equal(m.async, false)
+  m.async = true
+  t.equal(m.async, true)
+  m.async = 'banana'
+  t.equal(m.async, true)
+  m.async = false
+  t.equal(m.async, true, 'cannot make an async stream sync')
+  const d1 = new MP({ encoding: 'utf8' })
+  const d2 = new MP({ encoding: 'utf8' })
+
+  m.pipe(d1)
+  m.write('hello, ')
+  m.pipe(d2)
+  setTimeout(() => {
+    m.write('world')
+    m.end()
+  })
+  return Promise.all([d1.concat(), d2.concat()])
+    .then(result => t.strictSame(result, ['hello, world', 'hello, world']))
+})
+
+t.test('data event', t => {
+  const m = new MP({ encoding: 'utf8', async: true })
+  const d1 = new MP({ encoding: 'utf8' })
+
+  const out1 = []
+  m.on('data', c => out1.push(c))
+  m.write('hello, ')
+  const out2 = []
+  m.on('data', c => out2.push(c))
+  m.pipe(d1)
+  m.end('world!')
+  return d1.concat().then(res => {
+    t.equal(res, 'hello, world!')
+    t.equal(out1.join(''), 'hello, world!')
+    t.equal(out2.join(''), 'hello, world!')
+  })
+})
+
+t.test('data event split', t => {
+  const m = new MP({ encoding: 'utf8', async: true })
+  const d1 = new MP({ encoding: 'utf8' })
+
+  const out1 = []
+  m.on('data', c => out1.push(c))
+  m.write('hello, ')
+  const out2 = []
+  m.on('data', c => out2.push(c))
+  m.pipe(d1)
+  setTimeout(() => m.end('world!'))
+  return d1.concat().then(res => {
+    t.equal(res, 'hello, world!')
+    t.equal(out1.join(''), 'hello, world!')
+    t.equal(out2.join(''), 'hello, world!')
+  })
+})
+
+t.test('defer error event', t => {
+  const m = new MP()
+  try { m.emit('error', new Error('poop')) } catch (_) {}
+  m.async = true
+  let calledErrorHandler = false
+  m.on('error', er => {
+    t.equal(calledErrorHandler, false)
+    t.match(er, { message: 'poop' })
+    calledErrorHandler = true
+    t.end()
+  })
+  t.equal(calledErrorHandler, false)
+})