61 lines
1.2 KiB
JavaScript
61 lines
1.2 KiB
JavaScript
|
'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;
|