feature: add unpipe(dest) method

PR-URL: #39
Credit: @isaacs
Close: #39
Reviewed-by: @isaacs

Author: isaacs

README.md

@@ -17,9 +17,6 @@ from this stream via `'data'` events or by calling `pipe()` into some other
 stream.  Calling `read()` requires the buffer to be flattened in some
 cases, which requires copying memory.
 
-There is also no `unpipe()` method.  Once you start piping, there is no
-stopping it!
-
 If you set `objectMode: true` in the options, then whatever is written will
 be emitted.  Otherwise, it'll do a minimal amount of Buffer copying to
 ensure proper Streams semantics when `read(n)` is called.
@@ -383,6 +380,10 @@ mp.end('bar')
   by default if you write() something other than a string or Buffer at any
   point.  Setting `objectMode: true` will prevent setting any encoding
   value.
+* `async` Defaults to `false`.  Set to `true` to defer data
+  emission until next tick.  This reduces performance slightly,
+  but makes Minipass streams use timing behavior closer to Node
+  core streams.  See [Timing](#timing) for more details.
 
 ### API
 
@@ -404,9 +405,12 @@ streams.
   from being emitted for empty streams until the stream is resumed.
 * `resume()` - Resume the stream.  If there's data in the buffer, it is all
   discarded.  Any buffered events are immediately emitted.
-* `pipe(dest)` - Send all output to the stream provided.  There is no way
-  to unpipe.  When data is emitted, it is immediately written to any and
-  all pipe destinations.
+* `pipe(dest)` - Send all output to the stream provided.  When
+  data is emitted, it is immediately written to any and all pipe
+  destinations.  (Or written on next tick in `async` mode.)
+* `unpipe(dest)` - Stop piping to the destination stream.  This
+  is immediate, meaning that any asynchronously queued data will
+  _not_ make it to the destination when running in `async` mode.
 * `on(ev, fn)`, `emit(ev, fn)` - Minipass streams are EventEmitters.  Some
   events are given special treatment, however.  (See below under "events".)
 * `promise()` - Returns a Promise that resolves when the stream emits

index.js

@@ -63,10 +63,13 @@ class Pipe {
     this.ondrain = () => src[RESUME]()
     dest.on('drain', this.ondrain)
   }
+  unpipe () {
+    this.dest.removeListener('drain', this.ondrain)
+  }
   end () {
+    this.unpipe()
     if (this.opts.end)
       this.dest.end()
-    this.dest.removeListener('drain', this.ondrain)
   }
 }
 
@@ -377,6 +380,14 @@ module.exports = class Minipass extends Stream {
     return dest
   }
 
+  unpipe (dest) {
+    const p = this.pipes.find(p => p.dest === dest)
+    if (p) {
+      this.pipes.splice(this.pipes.indexOf(p), 1)
+      p.unpipe()
+    }
+  }
+
   addListener (ev, fn) {
     return this.on(ev, fn)
   }

test/unpipe.js

@@ -0,0 +1,48 @@
+const t = require('tap')
+const Minipass = require('../')
+
+const src = new Minipass({ encoding: 'utf8' })
+const dest = new Minipass({ encoding: 'utf8' })
+const dest2 = new Minipass({ encoding: 'utf8' })
+const destOut = []
+const dest2Out = []
+const srcOut = []
+
+src.pipe(dest)
+src.pipe(dest2)
+
+dest.on('data', c => destOut.push(c))
+dest2.on('data', c => dest2Out.push(c))
+src.on('data', c => srcOut.push(c))
+
+src.write('hello')
+t.strictSame(destOut, ['hello'])
+t.strictSame(dest2Out, ['hello'])
+t.strictSame(srcOut, ['hello'])
+
+t.match(src.pipes, [
+  { dest },
+  { dest: dest2 },
+])
+
+src.unpipe(dest)
+
+t.match(src.pipes, [
+  { dest: dest2 },
+])
+
+src.unpipe(dest) // no-op
+t.match(src.pipes, [
+  { dest: dest2 },
+])
+
+
+src.write('world')
+t.strictSame(destOut, ['hello'])
+t.strictSame(dest2Out, ['hello', 'world'])
+t.strictSame(srcOut, ['hello', 'world'])
+
+src.end()
+t.equal(dest.emittedEnd, false)
+t.equal(src.emittedEnd, true)
+t.equal(dest2.emittedEnd, true)