node pipe流時雙工流中當前流如何獲取到上一個流已經結束

node pipe流時雙工流中當前流如何獲取到上一個流已經結束。如下代碼中在SetIndexHtmlDuplexStream. prototype._write中我無法知道上一個可讀流中何時結束，所以無法知道在什么時候 this.push(null) 造成無法觸發(fā)currentCreateWriteStream finish事件。麻煩知道的大神指點一條明路

util.inherits(SetIndexHtmlDuplexStream, stream.Duplex);

function SetIndexHtmlDuplexStream(program, opt) {
    stream.Duplex.call(this, {
        objectMode: true,
        // allowHalfOpen: false
    });
    this.program = program;
}

SetIndexHtmlDuplexStream.prototype._write = function(chunk, encoding, callback) {

    let {
        output,
        name
    } = this.program;
    chunk = chunk.toString()
        .replace(/(src\=\"\.\/)(.+?)(\")/, '$1' + output + '$3') //設置rollup.config.js input 入口文件名字
        .replace(/(new\s)(.+?)(\(\))/, '$1' + name + '$3') //設置rollup.config.js input 入口文件名字

    this.push(chunk);

    callback();

}

SetIndexHtmlDuplexStream.prototype._read = function(size) {
}

let currentCreateReadStream = fs.createReadStream(inputPath);
let currentCreateWriteStream = fs.createWriteStream(outputPath);
const setIndexHtmlDuplexStream = new SetIndexHtmlDuplexStream(program);


currentCreateReadStream.pipe(setIndexHtmlDuplexStream).pipe(currentCreateWriteStream);

currentCreateWriteStream.on('finish', function() {
        console.log('   \x1b[36mcreate\x1b[0m : ' + path.relative(cwd, outputPath));
});

回答

編輯回答

不將就

查看了源碼發(fā)現(xiàn)pipe其實是做了上一個流結束會觸發(fā)下一個流結束的操作。

var endFn = doEnd ? onend : cleanup;
    if (state.endEmitted)
        process.nextTick(endFn);
    else
        src.once('end', endFn);

    dest.on('unpipe', onunpipe);

    function onunpipe(readable) {
        debug('onunpipe');
        if (readable === src) {
            cleanup();
        }
    }

    function onend() {
        debug('onend');
        dest.end();
    }
    function cleanup() {
        debug('cleanup');
        // cleanup event handlers once the pipe is broken
        dest.removeListener('close', onclose);
        dest.removeListener('finish', onfinish);
        dest.removeListener('drain', ondrain);
        dest.removeListener('error', onerror);
        dest.removeListener('unpipe', onunpipe);
        src.removeListener('end', onend);
        src.removeListener('end', cleanup);
        src.removeListener('data', ondata);

        cleanedUp = true;

        // if the reader is waiting for a drain event from this
        // specific writer, then it would cause it to never start
        // flowing again.
        // So, if this is awaiting a drain, then we just call it now.
        // If we don't know, then assume that we are waiting for one.
        if (state.awaitDrain &&
            (!dest._writableState || dest._writableState.needDrain))
            ondrain();
    }

但是不知道duplex抽什么風了。然后改用transiform就妥妥了

2018年3月12日 23:25