1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
|
'use strict'
var through2 = require('through2')
var inherits = require('inherits')
var nextTick = require('process-nextick-args')
var Ctor = through2.ctor()
function Cloneable (stream, opts) {
if (!(this instanceof Cloneable)) {
return new Cloneable(stream, opts)
}
var objectMode = stream._readableState.objectMode
this._original = stream
this._clonesCount = 1
opts = opts || {}
opts.objectMode = objectMode
Ctor.call(this, opts)
forwardDestroy(stream, this)
this.on('newListener', onData)
}
inherits(Cloneable, Ctor)
function onData (event, listener) {
if (event === 'data' || event === 'readable') {
this.removeListener('newListener', onData)
nextTick(clonePiped, this)
}
}
Cloneable.prototype.clone = function () {
if (!this._original) {
throw new Error('already started')
}
this._clonesCount++
// the events added by the clone should not count
// for starting the flow
this.removeListener('newListener', onData)
var clone = new Clone(this)
this.on('newListener', onData)
return clone
}
function forwardDestroy (src, dest) {
src.on('error', destroy)
src.on('close', destroy)
function destroy (err) {
dest.destroy(err)
}
}
function clonePiped (that) {
if (--that._clonesCount === 0 && !that._destroyed) {
that._original.pipe(that)
that._original = undefined
}
}
function Clone (parent, opts) {
if (!(this instanceof Clone)) {
return new Clone(parent, opts)
}
var objectMode = parent._readableState.objectMode
opts = opts || {}
opts.objectMode = objectMode
this.parent = parent
Ctor.call(this, opts)
forwardDestroy(this.parent, this)
parent.pipe(this)
// the events added by the clone should not count
// for starting the flow
// so we add the newListener handle after we are done
this.on('newListener', onDataClone)
}
function onDataClone (event, listener) {
// We start the flow once all clones are piped or destroyed
if (event === 'data' || event === 'readable' || event === 'close') {
nextTick(clonePiped, this.parent)
this.removeListener('newListener', onDataClone)
}
}
inherits(Clone, Ctor)
Clone.prototype.clone = function () {
return this.parent.clone()
}
Cloneable.isCloneable = function (stream) {
return stream instanceof Cloneable || stream instanceof Clone
}
module.exports = Cloneable
|
