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
|
'use strict';
var through = require('through2');
function forward(chunk, enc, cb) {
cb(null, chunk);
}
function toThrough(readable) {
var opts = {
objectMode: readable._readableState.objectMode,
highWaterMark: readable._readableState.highWaterMark,
};
function flush(cb) {
var self = this;
readable.on('readable', onReadable);
readable.on('end', cb);
function onReadable() {
var chunk;
while (chunk = readable.read()) {
self.push(chunk);
}
}
}
var wrapper = through(opts, forward, flush);
var shouldFlow = true;
wrapper.once('pipe', onPipe);
wrapper.on('newListener', onListener);
readable.on('error', wrapper.emit.bind(wrapper, 'error'));
function onListener(event) {
// Once we've seen the data or readable event, check if we need to flow
if (event === 'data' || event === 'readable') {
maybeFlow();
this.removeListener('newListener', onListener);
}
}
function onPipe() {
// If the wrapper is piped, disable flow
shouldFlow = false;
}
function maybeFlow() {
// If we need to flow, end the stream which triggers flush
if (shouldFlow) {
wrapper.end();
}
}
return wrapper;
}
module.exports = toThrough;
|
