Node.js + socket.io + several database queries needed for the result

Question

I need help.

I am trying to get around asynchronous programming of node.js and socket.io during the day. I understand that I need flow control, but I don't seem to understand how to implement it correctly.

I have a redis repository that has the modules stored in the set, let them say 'moda', 'modb' the instances of these modules are "moda: instances" and in "modb: instance" the properties of these instances are stored in "moda: instancea" and "modb: instanceb" as a hash.

I am trying to get the following json:

"moda": {"instancea": {"property1": "value1", "property2", "value2"}}, "modb": {"instanceb": {"property1": "value1"}} 

Can someone push me a little in the right direction?

Here is my current code:

 var io = require('socket.io').listen(2000); var redis = require('redis').createClient(); var http = require('http'); var async = require('async'); var step = require('step'); io.sockets.on('connection', function (socket) { var notifications = require('redis').createClient(); notifications.subscribe("notification"); notifications.on("message", function (channel, message) { socket.send(message); console.log(channel + ':' + message); }); socket.on('modules', function(params, callback) { var response = {}; async.series([ function (callback) { console.log('1>'); redis.smembers('modules', function (err, modules) { async.forEachSeries(modules, function(module, moduleCallback) { response[module] = {} redis.smembers(module + ':instances', function(err, instances) { async.forEachSeries(instances, function(instance, instanceCallback) { response[module][instance] = {} console.log('2>' + module + ':' +instance); instanceCallback(); }); moduleCallback(); }); }); callback(); }); }, function (callback) { console.log('3'); callback(); } ], function() { console.log(JSON.stringify(response)); }); }); }); 

Exit from this code:

  info - socket.io started debug - client authorized info - handshake authorized JMMn1I8aiOMGCMPOhC11 debug - setting request GET /socket.io/1/websocket/JMMn1I8aiOMGCMPOhC11 debug - set heartbeat interval for client JMMn1I8aiOMGCMPOhC11 debug - client authorized for debug - websocket writing 1:: 1> 3 {"moda":{}} 2>moda:instancea 2>moda:instanceb 2>modb:instancea 

Answer 1

The problem arises because forEachSeries requires an additional callback as the third parameter (called when all processing is complete). You cannot just put some code after forEachSeries hopes that it will be called after it completes.

Here is your code modified:

 var response = {}; async.series([ function (callback) { console.log('1>'); redis.smembers('modules', function (err, modules) { async.forEachSeries(modules, function(module, moduleCallback) { response[module] = {} redis.smembers(module + ':instances', function(err, instances) { async.forEachSeries(instances, function(instance, instanceCallback) { response[module][instance] = {} console.log('2>' + module + ':' +instance); instanceCallback(); }, moduleCallback ); }); }, callback ); }); }, function (callback) { console.log('3'); callback(); }], function() { console.log(JSON.stringify(response)); }); 

Notice how the callback and moduleCallback are used as the third parameter. Output:

 1> 2>moda:instancea 2>moda:instanceb 2>modb:instancea 3 {"moda":{"instancea":{},"instanceb":{}},"modb":{"instancea":{}}} 

I think this is what you expected.

Additional note: forEachSeries will process everything sequentially, the next operation waiting for the previous one to complete. This will result in a lot of calls to Redis. forEach must be much more efficient to use pipelining.

Answer 2

Take a look also at the concept of a promise, while promises you will configure the stream much more readable.

First you need to prepare the daily versions of the redis methods that you use:

 var promisify = require('deferred').promisify; var RedisClient = require('redis').RedisClient; RedisClient.prototype.psmembers = promisify(RedisClient.prototype.smembers); 

Then you can create your stream as:

 console.log('1>'); redis.psmembers('modules').map(function (module) { response[module] = {}; return redis.psmembers(module + ':instances').map(function (instance) { response[module][instance] = {}; console.log('2>' + module + ':' +instance); }); }).end(function () { console.log('3'); console.log(JSON.stringify(response)); }); 

Note: https://github.com/medikoo/deferred