node.js - Avoiding callback hell with multiple streams

Question

How can I avoid using a recursion like structure when I got several streams to open and I have to get an absolute end event to finish the logic.

var someArray = ['file1', 'file2', 'file3'];

someArray.forEach(function( file ) {
    fs
        .createReadStream( file )
        .pipe( /* do some stuff */ )
        .on('data', function( usageInfo ) {

            // done?

        });
}

I got several files I have to pipe through tp some processes. How can I setup an event that tells me when all of them are done? Currently what I'm getting is each end event individually.

I can absolutely start each stream at the same time. I just need to somehow collect the end?

I could invoke a function call for each end event and count it... that sounds hacky though?...

I feel like there is a way to do this with promises but I don't know how.

Answer 1

Use a counter:

var someArray = ['file1', 'file2', 'file3'];
var still_processing = someArray.length;

someArray.forEach(function( file ) {
    fs.createReadStream( file )
        .pipe( /* do some stuff */ )
        .on('end', function() {
            still_processing--;

            if (!still_processing) {
                // done
            }
        });
}

This is the basic mechanism. This control flow pattern is encapsulated by the async.parallel() function in async.js:

var someArray = ['file1', 'file2', 'file3'];
var streams_to_process = [];

someArray.forEach(function( file ) {
    streams_to_process.push(function(callback) {
        var result = "";
        fs.createReadStream( file )
            .pipe( /* do some stuff */ )
            .on('end', function() {
                callback(null, result);
            });
    });
});

async.parallel(streams_to_process, function(err, results) {
    // all done
});

Internally, async.parallel uses a counter captured in a closure to keep track of when all async processes (in this case the 'end' events) are done.

There are other libraries for this. Most promise libraries for example provide an .all() method that works the same way - internally keeping track of a counter value and firing the .then() callback when all is done.