Nodejs waits until all MongoDB calls end

Question

I read the stream of data from the CSV file in a row line by line and call findOne MongoDB for each line, how can I wait for all the mango calls from each line to complete before I run the next function?

I saw Promises can do this? But I find Promises extremely difficult to understand. And not one of the examples that I have found seems to cover what I'm trying to do.: /

var validProducts = []; fs.createReadStream(req.file.path) .pipe(csvStream) .on('error', function (err) { console.error(err); }) // loop through all rows .on('data', function (data) { if (data.size === 'a3') { ProductModel.findOne({ sku: data.sku }, function (err, product) { if (product !== null) { product.size = data.size; product.save(); validProducts.push(product); } }); } }); // on finish make call to other function socket.emit({ 'status': 'complete' }); otherFunction(validProducts); 

on('finish') or on('end') will only be called at the end of the data stream, and not after Monogo calls.

If I can use promises, can someone explain how?

Answer 1

You can use the Q API to make promises. There is an interesting feature that allows you to wait until the promises array is resolved. Here is an example of how you could solve your problem with Q.all :

 var validProducts = []; var promises = []; function handleData(data) { if (data.size === 'a3') { var deferred = Q.defer(); ProductModel.findOne({ sku: data.sku }, function (err, product) { if (err) { deferred.reject(new Error(err)); } if (product) { product.size = data.size; product.save(); deferred.resolve(product); validProducts.push(product); } }); promises.push(deferred.promise); } } function handleEnd() { Q.all(promises).done(function (values) { socket.emit({ 'status': 'complete' }); otherFunction(validProducts); }); } fs.createReadStream(req.file.path) .on('data', handleData) .on('end', handleEnd); 

Answer 2

Use pause / resume

  .on('data', function (data) { if (data.size === 'a3') { this.pause(); // pause it var stream = this; // save 'this' ProductModel.findOne({ sku: data.sku }, function (err, product) { if (product !== null) { product.size = data.size; product.save(); validProducts.push(product); } stream.resume(); //resume it }); } });