подождите, пока предыдущий поток будет пустым, прежде чем разрешить чтение

Question

Скажем, у меня есть файл, содержащий список целых чисел, по одному на строку. Я использую fs.createReadStream и передаю его в split (так что каждый кусок является целым числом). Затем я передаю это в дуплексный поток, который должен добавить числа и записать сумму по трубопроводу в fs.createWriteStream.

var fs = require('fs'); 
var stream = require('stream'); 
var split = require('split'); 

var addIntegers = new stream.Duplex(); 

addIntegers.sum = 0; 

addIntegers._read = function(size) { 
    this.push(this.sum + '\n'); 
} 

addIntegers._write = function(chunk, encoding, done) { 
    this.sum += +chunk; 
    done(); 
} 

fs.createReadStream('list-of-integers.txt') 
    .pipe(split()) 
    .pipe(addIntegers) 
    .pipe(fs.createWriteStream('sum.txt')); 

Когда я запускаю это, sum.txt просто получает постоянно заполняются нулями, и программа никогда не заканчивается (как и ожидалось). Как подождать, пока входной поток (split) будет пустым, прежде чем разрешить поток вывода (fs.createWriteStream) читать с addIntegers?

Answer 1

Я понял.

Я решил использовать поток Transform вместо (спасибо mscdex), потому что он имеет метод (_flush), который вызывается после того, как все записанные данные будут уничтожены. Рабочий код приведен ниже. Не забудьте указать npm i split :)

var fs = require('fs'); 
var stream = require('stream'); 
var split = require('split'); 

var addIntegers = new stream.Transform(); 

addIntegers.sum = 0; 

addIntegers._transform = function(chunk, encoding, done) { 
    this.sum += +chunk; 
    done(); 
} 

addIntegers._flush = function(done) { 
    this.push(this.sum + '\n'); 
} 

fs.createReadStream('list-of-integers.txt') 
    .pipe(split()) 
    .pipe(addIntegers) 
    .pipe(fs.createWriteStream('sum.txt'));