echo/rabbitmq-tutorials
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactored rpc_client.js

Browse Source
This commit is contained in:
Giuseppe Privitera 2015-08-06 17:42:29 +01:00
parent 38411ac07f
commit 1f64bb0b58
1 changed files with 20 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

51
javascript-nodejs/src/rpc_client.js
View File

@ -1,45 +1,34 @@
#!/usr/bin/env node #!/usr/bin/env node
var amqp = require('amqplib');
var when = require('when');
var conn = amqp.connect('amqp://localhost') var amqp = require('amqplib/callback_api');
conn.then(createChannel).then(null, console.warn);
function createChannel(conn) { var args = process.argv.slice(2);
return when(conn.createChannel().then(requestFib)).ensure(function() { conn.close(); });
if (args.length == 0) {
console.log("Usage: rpc_client.js num");
process.exit(1);
} }
function requestFib(ch) { amqp.connect('amqp://localhost', function(err, conn) {
var answer = when.defer(); conn.createChannel(function(err, ch) {
var correlationId = generateUuid(); ch.assertQueue('', {exclusive: true}, function(err, q) {
var corr = generateUuid();
var num = parseInt(args[0]);
function maybeAnswer(msg) { console.log(' [x] Requesting fib(%d)', num);
if (msg.properties.correlationId === correlationId) {
answer.resolve(msg.content.toString());
}
}
var ok = ch.assertQueue('', {exclusive: true}) ch.consume(q.queue, function(msg) {
.then(function(qok) { return qok.queue; }); console.log(' [.] Got %s', msg.content.toString());
setTimeout(function() { conn.close(); process.exit(0) }, 500);
}, {noAck: true});
ok = ok.then(function(queue) { ch.sendToQueue('rpc_queue',
return ch.consume(queue, maybeAnswer, {noAck: true}) new Buffer(num.toString()),
.then(function() { return queue; }); { correlationId: corr, replyTo: q.queue });
});
ok = ok.then(function(queue) {
console.log(' [x] Requesting fib(30)');
ch.sendToQueue('rpc_queue', new Buffer('30'), {
correlationId: correlationId, replyTo: queue
}); });
return answer.promise;
}); });
});
return ok.then(function(fibN) {
console.log(' [.] Got %d', fibN);
});
}
function generateUuid() { function generateUuid() {
return Math.random().toString() + Math.random().toString() + Math.random().toString(); return Math.random().toString() + Math.random().toString() + Math.random().toString();