warten Sie, bis der vorherige Stream leer ist, bevor Sie lesen können

Question

Angenommen, ich habe eine Datei mit einer Liste von ganzen Zahlen, eine pro Zeile. Ich verwende fs.createReadStream und pipe das in split (so dass jeder Chunk eine ganze Zahl ist). Dann piped ich das in einen Duplex-Strom, der die Zahlen hinzufügen soll, und schreibe die Summe durch Piping in 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')); 

Als ich dies ausführen, wird mit Nullen und das Programm nie beendet sum.txt nur ständig gefüllt (wie erwartet). Wie warte ich darauf, dass der Eingabestrom (split) leer ist, bevor der Ausgabestrom (fs.createWriteStream) von addIntegers gelesen werden kann?

Answer 1

Ich fand es heraus.

Ich entschied mich stattdessen einen Transform Stream zu verwenden (danke mscdex), weil es eine Methode (_flush) hat, die aufgerufen wird, nachdem alle geschriebenen Daten verbraucht sind. Der Arbeitscode ist unten. Vergessen Sie nicht, 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'));