1 | var pump = require('pump')
|
2 | var util = require('util')
|
3 | var duplexify = require('duplexify')
|
4 |
|
5 | var toArray = function(arguments) {
|
6 | if (!arguments.length) return []
|
7 | return Array.isArray(arguments[0]) ? arguments[0] : Array.prototype.slice.call(arguments)
|
8 | }
|
9 |
|
10 | var define = function(opts) {
|
11 | var Pumpify = function() {
|
12 | var streams = toArray(arguments)
|
13 |
|
14 | if (!(this instanceof Pumpify)) return new Pumpify(streams)
|
15 | duplexify.call(this, null, null, opts)
|
16 |
|
17 | if (streams.length) this.setPipeline(streams)
|
18 | }
|
19 |
|
20 | util.inherits(Pumpify, duplexify)
|
21 |
|
22 | Pumpify.prototype.setPipeline = function() {
|
23 | var streams = toArray(arguments)
|
24 | var self = this
|
25 |
|
26 | var w = streams[0]
|
27 | var r = streams[streams.length-1]
|
28 |
|
29 | r = r.readable ? r : null
|
30 | w = w.writable ? w : null
|
31 |
|
32 | var onprefinish = function(cb) {
|
33 | if (r._writableState.ended) return cb()
|
34 | r.on('finish', cb)
|
35 | }
|
36 |
|
37 | var onclose = function() {
|
38 | streams[0].emit('error', new Error('stream was destroyed'))
|
39 | }
|
40 |
|
41 | if (r && r._writableState) this.on('prefinish', onprefinish)
|
42 | this.on('close', onclose)
|
43 | pump(streams, function(err) {
|
44 | self.removeListener('close', onclose)
|
45 | self.destroy(err)
|
46 | })
|
47 |
|
48 | if (this.destroyed) return onclose()
|
49 | this.setWritable(w)
|
50 | this.setReadable(r)
|
51 | }
|
52 |
|
53 | return Pumpify
|
54 | }
|
55 |
|
56 | module.exports = define({destroy:false})
|
57 | module.exports.obj = define({destroy:false, objectMode:true, highWaterMark:16}) |
\ | No newline at end of file |