From 8d71478b31c5563f799b995537d8fbcff337b3a8 Mon Sep 17 00:00:00 2001 From: Michele Rastelli Date: Fri, 5 Jun 2015 13:19:31 +0200 Subject: [PATCH 01/23] throw errors --- index.js | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/index.js b/index.js index 44c965f..e47d04a 100644 --- a/index.js +++ b/index.js @@ -70,6 +70,10 @@ rpc.prototype._connect = function(cb) { this.__impl_options ); + this.__conn.on('error', function (err) { + throw err; + }); + this.__conn.on('ready', function() { debug("connected to " + $this.__conn.serverProperties.product); var cbs = $this.__connCbs; From 6fd94d383dc4e6f7aa313402a232635f4e680c23 Mon Sep 17 00:00:00 2001 From: Michele Rastelli Date: Mon, 8 Jun 2015 12:56:46 +0200 Subject: [PATCH 02/23] "amqp": "latest" --- package.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/package.json b/package.json index ff0a711..62ec172 100644 --- a/package.json +++ b/package.json @@ -28,7 +28,7 @@ } ], "dependencies": { - "amqp": "0.2.0", + "amqp": "latest", "node-uuid": "*", "debug": "~0.7.2" }, From e92513dc8ac40b7bc9cae3cc701371df6a7d2144 Mon Sep 17 00:00:00 2001 From: Michele Rastelli Date: Mon, 8 Jun 2015 15:41:56 +0200 Subject: [PATCH 03/23] .call renamed to .rpcCall --- index.js | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/index.js b/index.js index e47d04a..4d4b32c 100644 --- a/index.js +++ b/index.js @@ -207,7 +207,7 @@ rpc.prototype.__onResult = function(message, headers, deliveryInfo) { * @param {object} options advanced options of amqp */ -rpc.prototype.call = function(cmd, params, cb, context, options) { +rpc.prototype.rpcCall = function(cmd, params, cb, context, options) { debug('call()', cmd); var $this = this; @@ -409,7 +409,7 @@ rpc.prototype.callBroadcast = function(cmd, params, options) { options || (options = {}); options.broadcast = true; options.autoDeleteCallback = options.ttl ? false : true; - var corr_id = this.call.call(this, cmd, params, options.onResponse, options.context, options); + var corr_id = this.rpcCall.call(this, cmd, params, options.onResponse, options.context, options); if(options.ttl) { setTimeout(function() { //release cb From d05bcd3a495309958428dec9c46d7deb1e4d1566 Mon Sep 17 00:00:00 2001 From: Michele Rastelli Date: Tue, 9 Jun 2015 15:05:12 +0200 Subject: [PATCH 04/23] order arguments changed in rpcCall --- example/round-robin/client.js | 30 ++++++++++++++++++++++-------- example/round-robin/server.js | 24 ++++++++++++++++++++++-- index.js | 9 +++++---- package.json | 2 +- 4 files changed, 50 insertions(+), 15 deletions(-) diff --git a/example/round-robin/client.js b/example/round-robin/client.js index ee59de2..50984cc 100644 --- a/example/round-robin/client.js +++ b/example/round-robin/client.js @@ -1,18 +1,32 @@ +Object.defineProperty(Error, 'fromJSON', { + value: function (other) { + var err = new Error(); + Object.getOwnPropertyNames(other).forEach(function (key) { + err[key] = other[key]; + }); + return err; + }, + configurable: true +}); var rpc = require('../../index').factory({ - conn_options: { url: "amqp://guest:guest@localhost:5672", heartbeat: 10 } + conn_options: {url: "amqp://tdev1:mq4tdev1@127.0.0.1:5672", heartbeat: 10, confirm: true} }); -rpc.call('inc', 5, function() { - console.log('results of inc:', arguments); //output: [6,4,7] +rpc.rpcCall('inc', 5, null, null, function () { + console.log('results of inc:', arguments); //output: [6,4,7] }); -rpc.call('say.Hello', { name: 'John' }, function(msg) { - console.log('results of say.Hello:', msg); //output: Hello John! +rpc.rpcCall('say.Hello', {name: 'John'}, null, null, function (msg) { + console.log('results of say.Hello:', msg); //output: Hello John! }); -rpc.call('withoutCB', {}, function(msg) { - console.log('withoutCB results:', msg); //output: please run function without cb parameter +rpc.rpcCall('withoutCB', {}, null, null, function (msg) { + console.log('withoutCB results:', msg); //output: please run function without cb parameter }); -rpc.call('withoutCB', {}); //output message on server side console +//rpc.rpcCall('withoutCB', {}); //output message on server side console + +//rpc.rpcCall('errorFn', null, null, null, function (err, succ) { +// throw Error.fromJSON(err); +//}); diff --git a/example/round-robin/server.js b/example/round-robin/server.js index 8a1993a..5c6b3fc 100644 --- a/example/round-robin/server.js +++ b/example/round-robin/server.js @@ -1,6 +1,19 @@ +Object.defineProperty(Error.prototype, 'toJSON', { + value: function () { + var alt = {}; + + Object.getOwnPropertyNames(this).forEach(function (key) { + alt[key] = this[key]; + }, this); + + return alt; + }, + configurable: true +}); + var rpc = require('../../index').factory({ - conn_options: { url: "amqp://guest:guest@localhost:5672", heartbeat: 10 } + conn_options: {url: "amqp://tdev1:mq4tdev1@127.0.0.1:5672", heartbeat: 10 } }); @@ -11,7 +24,9 @@ rpc.on('inc', function(param, cb){ }); rpc.on('say.*', function(param, cb, inf){ - + console.log(param); + console.log(inf); + console.log(arguments); var arr = inf.cmd.split('.'); var name = (param && param.name) ? param.name : 'world'; @@ -30,3 +45,8 @@ rpc.on('withoutCB', function(param, cb, inf) { } }); + +rpc.on('errorFn', function (param, cb) { + cb(new Error("errorFn"), null); +}); + diff --git a/index.js b/index.js index 4d4b32c..b82551c 100644 --- a/index.js +++ b/index.js @@ -202,12 +202,13 @@ rpc.prototype.__onResult = function(message, headers, deliveryInfo) { * call a remote command * @param {string} cmd command name * @param {Buffer|Object|String}params parameters of command - * @param {function} cb callback - * @param {object} context context of callback * @param {object} options advanced options of amqp + * @param {object} context context of callback + * @param {function} cb callback + * */ -rpc.prototype.rpcCall = function(cmd, params, cb, context, options) { +rpc.prototype.rpcCall = function(cmd, params, options, context, cb) { debug('call()', cmd); var $this = this; @@ -409,7 +410,7 @@ rpc.prototype.callBroadcast = function(cmd, params, options) { options || (options = {}); options.broadcast = true; options.autoDeleteCallback = options.ttl ? false : true; - var corr_id = this.rpcCall.call(this, cmd, params, options.onResponse, options.context, options); + var corr_id = this.rpcCall.call(this, cmd, params, options, options.context, options.onResponse); if(options.ttl) { setTimeout(function() { //release cb diff --git a/package.json b/package.json index 62ec172..dd286bc 100644 --- a/package.json +++ b/package.json @@ -5,7 +5,7 @@ "amqp", "rpc" ], - "version": "0.0.8", + "version": "0.1.8", "preferGlobal": true, "author": { "name": "Eugene Demchenko" From 467d8f2f6c150159c2e60f6dd5bf1f39807c77cd Mon Sep 17 00:00:00 2001 From: Michele Rastelli Date: Tue, 9 Jun 2015 16:23:35 +0200 Subject: [PATCH 05/23] server handlers to handle messages in parallel and in sequence: .on and .onParallel --- example/round-robin/client.js | 6 ++ example/round-robin/server.js | 8 ++- index.js | 131 ++++++++++++++++++++++++---------- 3 files changed, 106 insertions(+), 39 deletions(-) diff --git a/example/round-robin/client.js b/example/round-robin/client.js index 50984cc..6d5d6c8 100644 --- a/example/round-robin/client.js +++ b/example/round-robin/client.js @@ -30,3 +30,9 @@ rpc.rpcCall('withoutCB', {}, null, null, function (msg) { //rpc.rpcCall('errorFn', null, null, null, function (err, succ) { // throw Error.fromJSON(err); //}); + +rpc.rpcCall('waitsTooMuch', null, null, console, console.log); +rpc.rpcCall('waitsTooMuch', null, null, console, console.log); + +//rpc.rpcCall('waitsTooMuch', null, {"ttl": "2000"}, console, console.log); +//rpc.rpcCall('waitsTooMuch', null, {"expiration": "2000"}, console, console.log); diff --git a/example/round-robin/server.js b/example/round-robin/server.js index 5c6b3fc..b73cf73 100644 --- a/example/round-robin/server.js +++ b/example/round-robin/server.js @@ -24,9 +24,6 @@ rpc.on('inc', function(param, cb){ }); rpc.on('say.*', function(param, cb, inf){ - console.log(param); - console.log(inf); - console.log(arguments); var arr = inf.cmd.split('.'); var name = (param && param.name) ? param.name : 'world'; @@ -50,3 +47,8 @@ rpc.on('errorFn', function (param, cb) { cb(new Error("errorFn"), null); }); +rpc.on('waitsTooMuch', function(param, cb){ + console.log("waitsTooMuch"); + //cb("waitsTooMuch OK!"); + setTimeout(cb.bind(null, "waitsTooMuch OK!"), 5000); +}); diff --git a/index.js b/index.js index b82551c..bcfe3ac 100644 --- a/index.js +++ b/index.js @@ -292,54 +292,113 @@ rpc.prototype.rpcCall = function(cmd, params, options, context, cb) { */ -rpc.prototype.on = function(cmd, cb, context, options) { - debug('on(), routingKey=%s', cmd); - if(this.__cmds[ cmd ]) return false; - options || (options = {}); +rpc.prototype.onParallel = function(cmd, cb, context, options) { + debug('on(), routingKey=%s', cmd); + if(this.__cmds[ cmd ]) return false; + options || (options = {}); - var $this = this; + var $this = this; - this._connect(function() { + this._connect(function() { - $this.__conn.queue(options.queueName || cmd, function(queue) { - $this.__cmds[ cmd ] = { queue: queue }; - queue.subscribe(function(message, d, headers, deliveryInfo) { + $this.__conn.queue(options.queueName || cmd, function(queue) { + $this.__cmds[ cmd ] = { queue: queue }; + queue.subscribe(function(message, d, headers, deliveryInfo) { - var cmdInfo = { - cmd: deliveryInfo.routingKey, - exchange: deliveryInfo.exchange, - contentType: deliveryInfo.contentType, - size: deliveryInfo.size - }; + var cmdInfo = { + cmd: deliveryInfo.routingKey, + exchange: deliveryInfo.exchange, + contentType: deliveryInfo.contentType, + size: deliveryInfo.size + }; - if(deliveryInfo.correlationId && deliveryInfo.replyTo ) { + if(deliveryInfo.correlationId && deliveryInfo.replyTo ) { - return cb.call(context, message, function(err, data) { + return cb.call(context, message, function(err, data) { + var options = { + correlationId: deliveryInfo.correlationId + } - var options = { - correlationId: deliveryInfo.correlationId - } + $this.__exchange.publish( + deliveryInfo.replyTo, + Array.prototype.slice.call(arguments), + options + ); + }, cmdInfo); + } + else + return cb.call(context, message, null, cmdInfo); + }); - $this.__exchange.publish( - deliveryInfo.replyTo, - Array.prototype.slice.call(arguments), - options - ); - }, cmdInfo); - } - else - return cb.call(context, message, null, cmdInfo); - }); + $this._makeExchange(function(){ + queue.bind($this.__exchange, cmd); + }); - $this._makeExchange(function(){ - queue.bind($this.__exchange, cmd); - }); + }); + }); - }); - }); + return true; +} - return true; +/** + * add new command handler, it handles a message per time, at the end it sends the ack to rabbitMQ + * @param cmd + * @param cb + * @param context + * @param options + * @returns {boolean} + */ +rpc.prototype.on = function(cmd, cb, context, options) { + debug('on(), routingKey=%s', cmd); + if(this.__cmds[ cmd ]) return false; + options || (options = {}); + + var $this = this; + + this._connect(function() { + + $this.__conn.queue(options.queueName || cmd, function(queue) { + $this.__cmds[ cmd ] = { queue: queue }; + queue.subscribe({ + ack: true + }, function(message, d, headers, deliveryInfo) { + + var cmdInfo = { + cmd: deliveryInfo.routingKey, + exchange: deliveryInfo.exchange, + contentType: deliveryInfo.contentType, + size: deliveryInfo.size + }; + + if(deliveryInfo.correlationId && deliveryInfo.replyTo ) { + + return cb.call(context, message, function(err, data) { + queue.shift(); + var options = { + correlationId: deliveryInfo.correlationId + } + + $this.__exchange.publish( + deliveryInfo.replyTo, + Array.prototype.slice.call(arguments), + options + ); + }, cmdInfo); + } + else + return cb.call(context, message, null, cmdInfo); + }); + + $this._makeExchange(function(){ + queue.bind($this.__exchange, cmd); + }); + + }); + }); + + + return true; } /** From 656a058acaa2decdaf7446d3672fe2af7cb05cdf Mon Sep 17 00:00:00 2001 From: Michele Rastelli Date: Tue, 9 Jun 2015 16:25:54 +0200 Subject: [PATCH 06/23] test message expiration --- example/round-robin/client.js | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/example/round-robin/client.js b/example/round-robin/client.js index 6d5d6c8..6d7640e 100644 --- a/example/round-robin/client.js +++ b/example/round-robin/client.js @@ -31,8 +31,8 @@ rpc.rpcCall('withoutCB', {}, null, null, function (msg) { // throw Error.fromJSON(err); //}); -rpc.rpcCall('waitsTooMuch', null, null, console, console.log); -rpc.rpcCall('waitsTooMuch', null, null, console, console.log); +//rpc.rpcCall('waitsTooMuch', null, null, console, console.log); +//rpc.rpcCall('waitsTooMuch', null, null, console, console.log); -//rpc.rpcCall('waitsTooMuch', null, {"ttl": "2000"}, console, console.log); -//rpc.rpcCall('waitsTooMuch', null, {"expiration": "2000"}, console, console.log); +rpc.rpcCall('waitsTooMuch', null, {"expiration": "3000"}, console, console.log); +rpc.rpcCall('waitsTooMuch', null, {"expiration": "3000"}, console, console.log); From 5c2492d78670ef9af4b24a995191a4ec8772ba72 Mon Sep 17 00:00:00 2001 From: Michele Rastelli Date: Fri, 19 Jun 2015 16:43:13 +0200 Subject: [PATCH 07/23] changed the order of parameters (.on and .onParallel) --- index.js | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/index.js b/index.js index bcfe3ac..94c1237 100644 --- a/index.js +++ b/index.js @@ -292,7 +292,7 @@ rpc.prototype.rpcCall = function(cmd, params, options, context, cb) { */ -rpc.prototype.onParallel = function(cmd, cb, context, options) { +rpc.prototype.onParallel = function(cmd, cb, options, context) { debug('on(), routingKey=%s', cmd); if(this.__cmds[ cmd ]) return false; options || (options = {}); @@ -349,7 +349,7 @@ rpc.prototype.onParallel = function(cmd, cb, context, options) { * @param options * @returns {boolean} */ -rpc.prototype.on = function(cmd, cb, context, options) { +rpc.prototype.on = function(cmd, cb, options, context) { debug('on(), routingKey=%s', cmd); if(this.__cmds[ cmd ]) return false; options || (options = {}); From 30514a014031455a4b29150af3b22021a037e804 Mon Sep 17 00:00:00 2001 From: Michele Rastelli Date: Fri, 10 Jul 2015 21:55:53 +0200 Subject: [PATCH 08/23] unique uuid-based queues name --- index.js | 4 +--- package.json | 4 ++-- 2 files changed, 3 insertions(+), 5 deletions(-) diff --git a/index.js b/index.js index 94c1237..1d85a4e 100644 --- a/index.js +++ b/index.js @@ -1,5 +1,3 @@ - - var amqp = require('amqp'); var uuid = require('node-uuid').v4; var os = require('os'); @@ -35,7 +33,7 @@ function rpc(opt) { */ rpc.prototype.generateQueueName = function(type) { - return /*'njsListener:' +*/ os.hostname() + ':pid' + process.pid + ':' + type; + return uuid() + ':' + os.hostname() + ':pid' + process.pid + ':' + type; } diff --git a/package.json b/package.json index dd286bc..6a9e439 100644 --- a/package.json +++ b/package.json @@ -29,8 +29,8 @@ ], "dependencies": { "amqp": "latest", - "node-uuid": "*", - "debug": "~0.7.2" + "debug": "~0.7.2", + "node-uuid": "^1.4.3" }, "readme": "\n#AMQP-RPC\n\nRPC library based on AMQP protocol.\nTested with RabbitMQ on the highload project.\n\n\n###Install RabitMQ\n\n apt-get install rabbitmq-server\n\n###Install library\n\n npm install amqp-rpc\n\n##round-robin\n\nExample: Call remote function.\nRun multiple servers.js for round-robin shared.\n\n\n###server.js example\n\n var rpc = require('amqp-rpc').factory({\n url: \"amqp://guest:guest@localhost:5672\"\n });\n\n\n rpc.on('inc', function(param, cb){\n var prevVal = param;\n var nextVal = param+2;\n cb(++param, prevVal, nextVal);\n });\n\n rpc.on('say.*', function(param, cb, inf){\n\n var arr = inf.cmd.split('.');\n\n var name = (param && param.name) ? param.name : 'world';\n\n cb(arr[1] + ' ' + name + '!');\n\n });\n\n rpc.on('withoutCB', function(param, cb, inf) {\n\n if(cb){\n cb('please run function without cb parameter')\n }\n else{\n console.log('this is function withoutCB');\n }\n\n });\n\n\n\n###client.js example\n\n var rpc = require('amqp-rpc').factory({\n url: \"amqp://guest:guest@localhost:5672\"\n });\n\n rpc.call('inc', 5, function() {\n console.log('results of inc:', arguments); //output: [6,4,7]\n });\n\n rpc.call('say.Hello', { name: 'John' }, function(msg) {\n console.log('results of say.Hello:', msg); //output: Hello John!\n });\n\n rpc.call('withoutCB', {}, function(msg) {\n console.log('withoutCB results:', msg); //output: please run function without cb parameter\n });\n\n rpc.call('withoutCB', {}); //output message on server side console\n\n\n##broadcast\n\nExample: Core receiving data from all workers.\nRun multiple worker.js for broadcast witness.\nThe core.js must be launched after all worker.js instances.\n\n###example/broadcast/worker.js\n\n var os = require('os');\n var worker_name = os.hostname() + ':' + process.pid;\n var counter = 0;\n\n var rpc = require('../../index').factory({\n url: \"amqp://guest:guest@localhost:5672\"\n });\n\n rpc.onBroadcast('getWorkerStat', function(params, cb) {\n if(params && params.type == 'fullStat') {\n cb(null, {\n pid: process.pid,\n hostname: os.hostname(),\n uptime: process.uptime(),\n counter: counter++\n });\n }\n else {\n cb(null, { counter: counter++ })\n }\n });\n\n###example/broadcast/core.js\n\n var rpc = require('../../index').factory({\n url: \"amqp://guest:guest@localhost:5672\"\n });\n\n var all_stats = {};\n\n //rpc.callBroadcast() is rpc.call() + waiting multiple responses\n //If remote handler without response data, you can use rpc.call() for initiate broadcast calls.\n\n rpc.callBroadcast(\n 'getWorkerStat',\n { type: 'fullStat'}, //request parameters\n { //call options\n ttl: 1000, //wait response time (1 seconds), after run onComplete\n onResponse: function(err, stat) { //callback on each worker response\n all_stats[ stat.hostname+':'+ stat.pid ] = stat;\n\n },\n onComplete: function() { //callback on ttl expired\n console.log('----------------------- WORKER STATISTICS ----------------------------------------');\n for(var worker in all_stats) {\n s = all_stats[worker];\n console.log(worker, '\\tuptime=', s.uptime.toFixed(2) + ' seconds', '\\tcounter=', s.counter);\n }\n }\n });\n\n\nResults for three workers:\n\n ----------------------- WORKER STATISTICS ----------------------------------------\n host1:2612 \tuptime= 2470.39 seconds \tcounter= 2\n host2:1615 \tuptime= 3723.53 seconds \tcounter= 8\n host2:2822 \tuptime= 2279.16 seconds \tcounter= 3\n\nEugene Demchenko aka Goldy skype demchenkoe email demchenkoev@gmail.com\n", "readmeFilename": "README.md", From 1dbece2de4232348fab7a2ff1693170379e17403 Mon Sep 17 00:00:00 2001 From: Michele Rastelli Date: Mon, 3 Aug 2015 12:12:55 +0200 Subject: [PATCH 09/23] rpc broadcast --- example/broadcast/core.js | 2 +- example/broadcast/worker.js | 13 +++++++------ index.js | 23 ----------------------- 3 files changed, 8 insertions(+), 30 deletions(-) diff --git a/example/broadcast/core.js b/example/broadcast/core.js index 70c65ab..f3ac873 100644 --- a/example/broadcast/core.js +++ b/example/broadcast/core.js @@ -1,5 +1,5 @@ var rpc = require('../../index').factory({ - url: "amqp://guest:guest@localhost:5672" + url: "amqp://tdev1:mq4tdev1@127.0.0.1:5672" }); var all_stats = {}; diff --git a/example/broadcast/worker.js b/example/broadcast/worker.js index f9a3c52..8070a88 100644 --- a/example/broadcast/worker.js +++ b/example/broadcast/worker.js @@ -3,11 +3,12 @@ var worker_name = os.hostname() + ':' + process.pid; var counter = 0; var rpc = require('../../index').factory({ - url: "amqp://guest:guest@localhost:5672" + url: "amqp://tdev1:mq4tdev1@127.0.0.1:5672" }); -rpc.onBroadcast('getWorkerStat', function(params, cb) { - if(params && params.type == 'fullStat') { +rpc.on('getWorkerStat', function (params, cb) { + console.log("getWorkerStat: " + worker_name); + if (params && params.type == 'fullStat') { cb(null, { pid: process.pid, hostname: os.hostname(), @@ -16,9 +17,9 @@ rpc.onBroadcast('getWorkerStat', function(params, cb) { }); } else { - cb(null, { counter: counter++ }) + cb(null, {counter: counter++}) } -}); +}, {queueName: "test-" + process.pid}); -rpc.call('log', { worker: worker_name, message: 'worker started' }); \ No newline at end of file +rpc.rpcCall('log', {worker: worker_name, message: 'worker started'}); \ No newline at end of file diff --git a/index.js b/index.js index 1d85a4e..668de83 100644 --- a/index.js +++ b/index.js @@ -479,29 +479,6 @@ rpc.prototype.callBroadcast = function(cmd, params, options) { } } -/** - * subscribe to broadcast commands - * @param {string} cmd - * @param {function} cb - * @param {object} context - */ - -rpc.prototype.onBroadcast = function (cmd, cb, context, options) { - - options || (options = {}); - options.queueName = this.generateQueueName('broadcast:q'+ (queueNo++) ); - return this.on.call(this, cmd, cb, context, options); -} - - -/** - * - * @type {Function} - */ - -rpc.prototype.offBroadcast = rpc.prototype.off; - - module.exports.amqpRPC = rpc; module.exports.factory = function(opt) { From 95ee9307b4cca2ba18e74b7d19e41a548045276a Mon Sep 17 00:00:00 2001 From: Michele Rastelli Date: Mon, 3 Aug 2015 14:13:40 +0200 Subject: [PATCH 10/23] bugfix multiple bindings --- example/round-robin/server.js | 4 +- index.js | 464 +++++++++++++++++++--------------- package.json | 2 +- 3 files changed, 269 insertions(+), 201 deletions(-) diff --git a/example/round-robin/server.js b/example/round-robin/server.js index b73cf73..4e748c2 100644 --- a/example/round-robin/server.js +++ b/example/round-robin/server.js @@ -17,11 +17,11 @@ var rpc = require('../../index').factory({ }); -rpc.on('inc', function(param, cb){ +rpc.on('zzttrr', function(param, cb){ var prevVal = param; var nextVal = param+2; cb(++param, prevVal, nextVal); -}); +}, {queueName: "test_inc"}); rpc.on('say.*', function(param, cb, inf){ var arr = inf.cmd.split('.'); diff --git a/index.js b/index.js index 668de83..b9f6636 100644 --- a/index.js +++ b/index.js @@ -1,20 +1,20 @@ var amqp = require('amqp'); var uuid = require('node-uuid').v4; -var os = require('os'); -var debug= require('debug')('amqp-rpc'); +var os = require('os'); +var debug = require('debug')('amqp-rpc'); var queueNo = 0; -function rpc(opt) { +function rpc(opt) { - if(!opt) opt = {}; + if (!opt) opt = {}; this.opt = opt; - this.__conn = opt.connection ? opt.connection : null; - this.__url = opt.url ? opt.url: 'amqp://guest:guest@localhost:5672'; - this.__exchange = opt.exchangeInstance ? opt.exchangeInstance : null; - this.__exchange_name = opt.exchange ? opt.exchange : 'rpc_exchange'; - this.__exchange_options = opt.exchange_options ? opt.exchange_options : {exclusive: false, autoDelete: true }; - this.__impl_options = opt.ipml_options || {defaultExchangeName: this.__exchange_name}; - this.__conn_options = opt.conn_options || {}; + this.__conn = opt.connection ? opt.connection : null; + this.__url = opt.url ? opt.url : 'amqp://guest:guest@localhost:5672'; + this.__exchange = opt.exchangeInstance ? opt.exchangeInstance : null; + this.__exchange_name = opt.exchange ? opt.exchange : 'rpc_exchange'; + this.__exchange_options = opt.exchange_options ? opt.exchange_options : {exclusive: false, autoDelete: true}; + this.__impl_options = opt.ipml_options || {defaultExchangeName: this.__exchange_name}; + this.__conn_options = opt.conn_options || {}; this.__results_queue = null; this.__results_queue_name = null; @@ -32,22 +32,23 @@ function rpc(opt) { * @returns {string} */ -rpc.prototype.generateQueueName = function(type) { +rpc.prototype.generateQueueName = function (type) { return uuid() + ':' + os.hostname() + ':pid' + process.pid + ':' + type; } -rpc.prototype._connect = function(cb) { +rpc.prototype._connect = function (cb) { - if(!cb) { - cb = function(){}; + if (!cb) { + cb = function () { + }; } - if(this.__conn) { + if (this.__conn) { - if(this.__connCbs.length > 0) { + if (this.__connCbs.length > 0) { - this.__connCbs.push(cb); + this.__connCbs.push(cb); return true; } @@ -61,10 +62,10 @@ rpc.prototype._connect = function(cb) { this.__connCbs.push(cb); var options = this.__conn_options; - if(!options.url && !options.host) options.url = this.__url; + if (!options.url && !options.host) options.url = this.__url; debug("createConnection options=", options, ', ipml_options=', this.__impl_options || {}); this.__conn = amqp.createConnection( - options, + options, this.__impl_options ); @@ -72,36 +73,37 @@ rpc.prototype._connect = function(cb) { throw err; }); - this.__conn.on('ready', function() { - debug("connected to " + $this.__conn.serverProperties.product); - var cbs = $this.__connCbs; - $this.__connCbs = []; + this.__conn.on('ready', function () { + debug("connected to " + $this.__conn.serverProperties.product); + var cbs = $this.__connCbs; + $this.__connCbs = []; - for(var i=0; i< cbs.length; i++) { - cbs[i]($this.__conn); - } + for (var i = 0; i < cbs.length; i++) { + cbs[i]($this.__conn); + } }); } /** * disconnect from MQ broker */ -rpc.prototype.disconnect = function() { +rpc.prototype.disconnect = function () { debug("disconnect()"); - if(!this.__conn) return; + if (!this.__conn) return; this.__conn.end(); this.__conn = null; } -rpc.prototype._makeExchange = function(cb) { +rpc.prototype._makeExchange = function (cb) { - if(!cb) { - cb = function(){}; + if (!cb) { + cb = function () { + }; } - if(this.__exchange) { + if (this.__exchange) { - if(this.__exchangeCbs.length > 0) { + if (this.__exchangeCbs.length > 0) { this.__exchangeCbs.push(cb); @@ -114,31 +116,32 @@ rpc.prototype._makeExchange = function(cb) { var $this = this; this.__exchangeCbs.push(cb); - /* - * Added option autoDelete=false. - * Otherwise we had an error in library node-amqp version > 0.1.7. - * Text of such error: "PRECONDITION_FAILED - cannot redeclare exchange '' in vhost '/' with different type, durable, internal or autodelete value" - */ - this.__exchange = this.__conn.exchange(this.__exchange_name, { autoDelete: false }, function(exchange) { + /* + * Added option autoDelete=false. + * Otherwise we had an error in library node-amqp version > 0.1.7. + * Text of such error: "PRECONDITION_FAILED - cannot redeclare exchange '' in vhost '/' with different type, durable, internal or autodelete value" + */ + this.__exchange = this.__conn.exchange(this.__exchange_name, {autoDelete: false}, function (exchange) { debug('Exchange ' + exchange.name + ' is open'); var cbs = $this.__exchangeCbs; $this.__exchangeCbs = []; - for(var i=0; i< cbs.length; i++) { + for (var i = 0; i < cbs.length; i++) { cbs[i]($this.__exchange); } }); } -rpc.prototype._makeResultsQueue = function(cb) { +rpc.prototype._makeResultsQueue = function (cb) { - if(!cb) { - cb = function(){}; + if (!cb) { + cb = function () { + }; } - if(this.__results_queue) { - if(this.__make_results_cb.length > 0) { + if (this.__results_queue) { + if (this.__make_results_cb.length > 0) { this.__make_results_cb.push(cb); return true; @@ -151,14 +154,14 @@ rpc.prototype._makeResultsQueue = function(cb) { this.__results_queue_name = this.generateQueueName('callback'); this.__make_results_cb.push(cb); - $this._makeExchange(function() { + $this._makeExchange(function () { $this.__results_queue = $this.__conn.queue( $this.__results_queue_name, $this.__exchange_options, - function(queue) { + function (queue) { debug('Callback queue ' + queue.name + ' is open'); - queue.subscribe(function() { + queue.subscribe(function () { $this.__onResult.apply($this, arguments); }); @@ -167,7 +170,7 @@ rpc.prototype._makeResultsQueue = function(cb) { var cbs = $this.__make_results_cb; $this.__make_results_cb = []; - for(var i=0; i Date: Mon, 3 Aug 2015 14:18:03 +0200 Subject: [PATCH 11/23] bugfix multiple bindings --- index.js | 22 ++++++++-------------- package.json | 2 +- 2 files changed, 9 insertions(+), 15 deletions(-) diff --git a/index.js b/index.js index b9f6636..12e3461 100644 --- a/index.js +++ b/index.js @@ -370,16 +370,13 @@ rpc.prototype.onParallel = function (cmd, cb, options, context) { }); $this._deleteBindings(queue, bindingsToDelete); - $this._makeExchange(function () { - queue.bind($this.__exchange, cmd); - }); - }); - } else { - $this._makeExchange(function () { - queue.bind($this.__exchange, cmd); }); } + $this._makeExchange(function () { + queue.bind($this.__exchange, cmd); + }); + }); }); @@ -447,16 +444,13 @@ rpc.prototype.on = function (cmd, cb, options, context) { }); $this._deleteBindings(queue, bindingsToDelete); - $this._makeExchange(function () { - queue.bind($this.__exchange, cmd); - }); - }); - } else { - $this._makeExchange(function () { - queue.bind($this.__exchange, cmd); }); } + $this._makeExchange(function () { + queue.bind($this.__exchange, cmd); + }); + }); }); diff --git a/package.json b/package.json index 6bab1ce..8181c50 100644 --- a/package.json +++ b/package.json @@ -5,7 +5,7 @@ "amqp", "rpc" ], - "version": "0.2.0", + "version": "0.2.1", "preferGlobal": true, "author": { "name": "Eugene Demchenko" From 574c58decfacd0a6bf460f002043e884bd6415f9 Mon Sep 17 00:00:00 2001 From: Michele Rastelli Date: Fri, 14 Aug 2015 10:40:07 +0200 Subject: [PATCH 12/23] print channels --- README.md | 8 +++++ index.js | 82 +++++++++++++++++++++++++++++++++++----------------- package.json | 2 +- 3 files changed, 64 insertions(+), 28 deletions(-) diff --git a/README.md b/README.md index 471c5b4..ef26b12 100644 --- a/README.md +++ b/README.md @@ -144,3 +144,11 @@ Results for three workers: host2:2822 uptime= 2279.16 seconds counter= 3 Eugene Demchenko aka Goldy skype demchenkoe email demchenkoev@gmail.com + + + +## Changelog + +0.2.2: + + * print channels fn diff --git a/index.js b/index.js index 12e3461..b7a04c6 100644 --- a/index.js +++ b/index.js @@ -327,11 +327,13 @@ rpc.prototype.onParallel = function (cmd, cb, options, context) { options || (options = {}); var $this = this; + options.queueName = options.queueName || cmd; this._connect(function () { - $this.__conn.queue(options.queueName || cmd, function (queue) { + $this.__conn.queue(options.queueName, function (queue) { $this.__cmds[cmd] = {queue: queue}; + $this.printChannels(options.queueName); queue.subscribe(function (message, d, headers, deliveryInfo) { var cmdInfo = { @@ -359,19 +361,17 @@ rpc.prototype.onParallel = function (cmd, cb, options, context) { return cb.call(context, message, null, cmdInfo); }); - if (options.queueName) { - $this._getCurrentBindings(options.queueName, "%2f", function (bindings) { - var bindingsToDelete = bindings.map(function (b) { - return {exchange: b.source, routing: b.routing_key}; - }).filter(function (b) { - return b.exchange && b.exchange !== ''; - }).filter(function (b) { - return !(b.exchange === $this.__exchange_name && b.routing === cmd); - }); - - $this._deleteBindings(queue, bindingsToDelete); + $this._getCurrentBindings(options.queueName, "%2f", function (bindings) { + var bindingsToDelete = bindings.map(function (b) { + return {exchange: b.source, routing: b.routing_key}; + }).filter(function (b) { + return b.exchange && b.exchange !== ''; + }).filter(function (b) { + return !(b.exchange === $this.__exchange_name && b.routing === cmd); }); - } + + $this._deleteBindings(queue, bindingsToDelete); + }); $this._makeExchange(function () { queue.bind($this.__exchange, cmd); @@ -399,10 +399,12 @@ rpc.prototype.on = function (cmd, cb, options, context) { var $this = this; - this._connect(function () { + options.queueName = options.queueName || cmd; - $this.__conn.queue(options.queueName || cmd, function (queue) { + this._connect(function () { + $this.__conn.queue(options.queueName, function (queue) { $this.__cmds[cmd] = {queue: queue}; + $this.printChannels(options.queueName); queue.subscribe({ ack: true }, function (message, d, headers, deliveryInfo) { @@ -433,19 +435,17 @@ rpc.prototype.on = function (cmd, cb, options, context) { return cb.call(context, message, null, cmdInfo); }); - if (options.queueName) { - $this._getCurrentBindings(options.queueName, "%2f", function (bindings) { - var bindingsToDelete = bindings.map(function (b) { - return {exchange: b.source, routing: b.routing_key}; - }).filter(function (b) { - return b.exchange && b.exchange !== ''; - }).filter(function (b) { - return !(b.exchange === $this.__exchange_name && b.routing === cmd); - }); - - $this._deleteBindings(queue, bindingsToDelete); + $this._getCurrentBindings(options.queueName, "%2f", function (bindings) { + var bindingsToDelete = bindings.map(function (b) { + return {exchange: b.source, routing: b.routing_key}; + }).filter(function (b) { + return b.exchange && b.exchange !== ''; + }).filter(function (b) { + return !(b.exchange === $this.__exchange_name && b.routing === cmd); }); - } + + $this._deleteBindings(queue, bindingsToDelete); + }); $this._makeExchange(function () { queue.bind($this.__exchange, cmd); @@ -541,6 +541,34 @@ rpc.prototype.callBroadcast = function (cmd, params, options) { } } +rpc.prototype.printChannels = function (queueName) { + var thisRabbit = this; + thisRabbit._getChannels(queueName, function (channels) { + if (channels.length > 0) { + console.log("Channels in [" + queueName + "] queue: "); + channels.forEach(function (v) { + console.log("\t" + v); + }); + } + }); +}; + +rpc.prototype._getChannels = function (queueName, cb) { + var routing = "rabbitmon"; + var message = {apiCall: "queues/" + "%2f" + "/" + queueName}; + this.rpcCall(routing, message, {expiration: "20000"}, null, function (err, res) { + if (err) { + console.error(err); + throw err; + } + var ipList = res.consumer_details.map(function (consumer) { + return consumer.channel_details.connection_name; + }); + + cb(ipList); + }); +}; + module.exports.amqpRPC = rpc; module.exports.factory = function (opt) { diff --git a/package.json b/package.json index 8181c50..9ecace6 100644 --- a/package.json +++ b/package.json @@ -5,7 +5,7 @@ "amqp", "rpc" ], - "version": "0.2.1", + "version": "0.2.2", "preferGlobal": true, "author": { "name": "Eugene Demchenko" From cdf4498d18fbac50f410f06a973c0e217ddf75b5 Mon Sep 17 00:00:00 2001 From: Michele Rastelli Date: Fri, 14 Aug 2015 10:58:30 +0200 Subject: [PATCH 13/23] print channels after subscribe --- index.js | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/index.js b/index.js index b7a04c6..dd59ccf 100644 --- a/index.js +++ b/index.js @@ -333,7 +333,6 @@ rpc.prototype.onParallel = function (cmd, cb, options, context) { $this.__conn.queue(options.queueName, function (queue) { $this.__cmds[cmd] = {queue: queue}; - $this.printChannels(options.queueName); queue.subscribe(function (message, d, headers, deliveryInfo) { var cmdInfo = { @@ -361,6 +360,8 @@ rpc.prototype.onParallel = function (cmd, cb, options, context) { return cb.call(context, message, null, cmdInfo); }); + $this.printChannels(options.queueName); + $this._getCurrentBindings(options.queueName, "%2f", function (bindings) { var bindingsToDelete = bindings.map(function (b) { return {exchange: b.source, routing: b.routing_key}; @@ -404,7 +405,7 @@ rpc.prototype.on = function (cmd, cb, options, context) { this._connect(function () { $this.__conn.queue(options.queueName, function (queue) { $this.__cmds[cmd] = {queue: queue}; - $this.printChannels(options.queueName); + queue.subscribe({ ack: true }, function (message, d, headers, deliveryInfo) { @@ -435,6 +436,8 @@ rpc.prototype.on = function (cmd, cb, options, context) { return cb.call(context, message, null, cmdInfo); }); + $this.printChannels(options.queueName); + $this._getCurrentBindings(options.queueName, "%2f", function (bindings) { var bindingsToDelete = bindings.map(function (b) { return {exchange: b.source, routing: b.routing_key}; From b1898883bbad7cd846102530ef88d1f31a670a79 Mon Sep 17 00:00:00 2001 From: Michele Rastelli Date: Fri, 14 Aug 2015 11:58:36 +0200 Subject: [PATCH 14/23] support for different exchanges --- example/round-robin/server.js | 23 ++++++++++++----------- index.js | 12 ++++++------ 2 files changed, 18 insertions(+), 17 deletions(-) diff --git a/example/round-robin/server.js b/example/round-robin/server.js index 4e748c2..1cba253 100644 --- a/example/round-robin/server.js +++ b/example/round-robin/server.js @@ -13,17 +13,18 @@ Object.defineProperty(Error.prototype, 'toJSON', { var rpc = require('../../index').factory({ - conn_options: {url: "amqp://tdev1:mq4tdev1@127.0.0.1:5672", heartbeat: 10 } + exchange: "uuu-test", exchange_options: {exclusive: false, autoDelete: true}, + conn_options: {url: "amqp://tdev1:mq4tdev1@127.0.0.1:5672", heartbeat: 10} }); rpc.on('zzttrr', function(param, cb){ var prevVal = param; - var nextVal = param+2; + var nextVal = param + 2; cb(++param, prevVal, nextVal); }, {queueName: "test_inc"}); -rpc.on('say.*', function(param, cb, inf){ +rpc.on('say.*', function (param, cb, inf) { var arr = inf.cmd.split('.'); var name = (param && param.name) ? param.name : 'world'; @@ -32,14 +33,14 @@ rpc.on('say.*', function(param, cb, inf){ }); -rpc.on('withoutCB', function(param, cb, inf) { +rpc.on('withoutCB', function (param, cb, inf) { - if(cb){ - cb('please run function without cb parameter') - } - else{ - console.log('this is function withoutCB'); - } + if (cb) { + cb('please run function without cb parameter') + } + else { + console.log('this is function withoutCB'); + } }); @@ -47,7 +48,7 @@ rpc.on('errorFn', function (param, cb) { cb(new Error("errorFn"), null); }); -rpc.on('waitsTooMuch', function(param, cb){ +rpc.on('waitsTooMuch', function (param, cb) { console.log("waitsTooMuch"); //cb("waitsTooMuch OK!"); setTimeout(cb.bind(null, "waitsTooMuch OK!"), 5000); diff --git a/index.js b/index.js index dd59ccf..5458c76 100644 --- a/index.js +++ b/index.js @@ -271,17 +271,15 @@ rpc.prototype.rpcCall = function (cmd, params, options, context, cb) { } rpc.prototype._getCurrentBindings = function (queueName, vhost, cb) { - //var protocol = server.ssl && server.ssl.enabled ? "amqps://" : "amqp://"; - //var url = protocol + server.login + ":" + server.password + "@" + server.host + ":" + server.port; - //var rpc = Rpc.factory({conn_options: {url: url}}); - + var myRpc = new rpc({exchange: "rpc_exchange", conn_options: {url: this.opt.conn_options.url}}); var routing = "rabbitmon"; var message = {apiCall: "queues/" + vhost + "/" + queueName + "/bindings"}; - this.rpcCall(routing, message, {expiration: "20000"}, null, function (err, res) { + myRpc.rpcCall(routing, message, {expiration: "20000"}, null, function (err, res) { if (err) { console.error(err); throw err; } + myRpc.disconnect(); cb(res); }); }; @@ -557,13 +555,15 @@ rpc.prototype.printChannels = function (queueName) { }; rpc.prototype._getChannels = function (queueName, cb) { + var myRpc = new rpc({exchange: "rpc_exchange", conn_options: {url: this.opt.conn_options.url}}); var routing = "rabbitmon"; var message = {apiCall: "queues/" + "%2f" + "/" + queueName}; - this.rpcCall(routing, message, {expiration: "20000"}, null, function (err, res) { + myRpc.rpcCall(routing, message, {expiration: "20000"}, null, function (err, res) { if (err) { console.error(err); throw err; } + myRpc.disconnect(); var ipList = res.consumer_details.map(function (consumer) { return consumer.channel_details.connection_name; }); From 1fa2a73591f5d342386b9c9a4bfd4e6d80c7dcbd Mon Sep 17 00:00:00 2001 From: Michele Rastelli Date: Fri, 14 Aug 2015 12:00:02 +0200 Subject: [PATCH 15/23] support for different exchanges --- README.md | 8 ++++++++ package.json | 2 +- 2 files changed, 9 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index ef26b12..9f6eafb 100644 --- a/README.md +++ b/README.md @@ -149,6 +149,14 @@ Eugene Demchenko aka Goldy skype demchenkoe email demchenkoev@gmail.com ## Changelog + +0.2.3: + + * support for other exchanges than rpc_exchange + + 0.2.2: * print channels fn + + diff --git a/package.json b/package.json index 9ecace6..27c6f09 100644 --- a/package.json +++ b/package.json @@ -5,7 +5,7 @@ "amqp", "rpc" ], - "version": "0.2.2", + "version": "0.2.3", "preferGlobal": true, "author": { "name": "Eugene Demchenko" From ac136a6a180f98ed96bc557efca349af9d5614c7 Mon Sep 17 00:00:00 2001 From: Michele Rastelli Date: Mon, 31 Aug 2015 16:14:21 +0000 Subject: [PATCH 16/23] rm dbg --- index.js | 12 ------------ package.json | 1 - 2 files changed, 13 deletions(-) diff --git a/index.js b/index.js index 5458c76..bff79bf 100644 --- a/index.js +++ b/index.js @@ -1,7 +1,6 @@ var amqp = require('amqp'); var uuid = require('node-uuid').v4; var os = require('os'); -var debug = require('debug')('amqp-rpc'); var queueNo = 0; function rpc(opt) { @@ -63,7 +62,6 @@ rpc.prototype._connect = function (cb) { this.__connCbs.push(cb); var options = this.__conn_options; if (!options.url && !options.host) options.url = this.__url; - debug("createConnection options=", options, ', ipml_options=', this.__impl_options || {}); this.__conn = amqp.createConnection( options, this.__impl_options @@ -74,7 +72,6 @@ rpc.prototype._connect = function (cb) { }); this.__conn.on('ready', function () { - debug("connected to " + $this.__conn.serverProperties.product); var cbs = $this.__connCbs; $this.__connCbs = []; @@ -88,7 +85,6 @@ rpc.prototype._connect = function (cb) { */ rpc.prototype.disconnect = function () { - debug("disconnect()"); if (!this.__conn) return; this.__conn.end(); this.__conn = null; @@ -122,7 +118,6 @@ rpc.prototype._makeExchange = function (cb) { * Text of such error: "PRECONDITION_FAILED - cannot redeclare exchange '' in vhost '/' with different type, durable, internal or autodelete value" */ this.__exchange = this.__conn.exchange(this.__exchange_name, {autoDelete: false}, function (exchange) { - debug('Exchange ' + exchange.name + ' is open'); var cbs = $this.__exchangeCbs; $this.__exchangeCbs = []; @@ -160,13 +155,11 @@ rpc.prototype._makeResultsQueue = function (cb) { $this.__results_queue_name, $this.__exchange_options, function (queue) { - debug('Callback queue ' + queue.name + ' is open'); queue.subscribe(function () { $this.__onResult.apply($this, arguments); }); queue.bind($this.__exchange, $this.__results_queue_name); - debug('Bind queue ' + queue.name + ' to exchange ' + $this.__exchange.name); var cbs = $this.__make_results_cb; $this.__make_results_cb = []; @@ -179,7 +172,6 @@ rpc.prototype._makeResultsQueue = function (cb) { } rpc.prototype.__onResult = function (message, headers, deliveryInfo) { - debug("__onResult()"); if (!this.__results_cb[deliveryInfo.correlationId]) return; var cb = this.__results_cb[deliveryInfo.correlationId]; @@ -210,7 +202,6 @@ rpc.prototype.__onResult = function (message, headers, deliveryInfo) { */ rpc.prototype.rpcCall = function (cmd, params, options, context, cb) { - debug('call()', cmd); var $this = this; if (!options) options = {}; @@ -320,7 +311,6 @@ rpc.prototype._deleteBindings = function (queue, bindingsToDelete) { rpc.prototype.onParallel = function (cmd, cb, options, context) { - debug('on(), routingKey=%s', cmd); if (this.__cmds[cmd]) return false; options || (options = {}); @@ -392,7 +382,6 @@ rpc.prototype.onParallel = function (cmd, cb, options, context) { * @returns {boolean} */ rpc.prototype.on = function (cmd, cb, options, context) { - debug('on(), routingKey=%s', cmd); if (this.__cmds[cmd]) return false; options || (options = {}); @@ -466,7 +455,6 @@ rpc.prototype.on = function (cmd, cb, options, context) { */ rpc.prototype.off = function (cmd) { - debug('off', cmd); if (!this.__cmds[cmd]) return false; var $this = this; diff --git a/package.json b/package.json index 27c6f09..6e741dc 100644 --- a/package.json +++ b/package.json @@ -29,7 +29,6 @@ ], "dependencies": { "amqp": "latest", - "debug": "~0.7.2", "node-uuid": "^1.4.3" }, "readme": "\n#AMQP-RPC\n\nRPC library based on AMQP protocol.\nTested with RabbitMQ on the highload project.\n\n\n###Install RabitMQ\n\n apt-get install rabbitmq-server\n\n###Install library\n\n npm install amqp-rpc\n\n##round-robin\n\nExample: Call remote function.\nRun multiple servers.js for round-robin shared.\n\n\n###server.js example\n\n var rpc = require('amqp-rpc').factory({\n url: \"amqp://guest:guest@localhost:5672\"\n });\n\n\n rpc.on('inc', function(param, cb){\n var prevVal = param;\n var nextVal = param+2;\n cb(++param, prevVal, nextVal);\n });\n\n rpc.on('say.*', function(param, cb, inf){\n\n var arr = inf.cmd.split('.');\n\n var name = (param && param.name) ? param.name : 'world';\n\n cb(arr[1] + ' ' + name + '!');\n\n });\n\n rpc.on('withoutCB', function(param, cb, inf) {\n\n if(cb){\n cb('please run function without cb parameter')\n }\n else{\n console.log('this is function withoutCB');\n }\n\n });\n\n\n\n###client.js example\n\n var rpc = require('amqp-rpc').factory({\n url: \"amqp://guest:guest@localhost:5672\"\n });\n\n rpc.call('inc', 5, function() {\n console.log('results of inc:', arguments); //output: [6,4,7]\n });\n\n rpc.call('say.Hello', { name: 'John' }, function(msg) {\n console.log('results of say.Hello:', msg); //output: Hello John!\n });\n\n rpc.call('withoutCB', {}, function(msg) {\n console.log('withoutCB results:', msg); //output: please run function without cb parameter\n });\n\n rpc.call('withoutCB', {}); //output message on server side console\n\n\n##broadcast\n\nExample: Core receiving data from all workers.\nRun multiple worker.js for broadcast witness.\nThe core.js must be launched after all worker.js instances.\n\n###example/broadcast/worker.js\n\n var os = require('os');\n var worker_name = os.hostname() + ':' + process.pid;\n var counter = 0;\n\n var rpc = require('../../index').factory({\n url: \"amqp://guest:guest@localhost:5672\"\n });\n\n rpc.onBroadcast('getWorkerStat', function(params, cb) {\n if(params && params.type == 'fullStat') {\n cb(null, {\n pid: process.pid,\n hostname: os.hostname(),\n uptime: process.uptime(),\n counter: counter++\n });\n }\n else {\n cb(null, { counter: counter++ })\n }\n });\n\n###example/broadcast/core.js\n\n var rpc = require('../../index').factory({\n url: \"amqp://guest:guest@localhost:5672\"\n });\n\n var all_stats = {};\n\n //rpc.callBroadcast() is rpc.call() + waiting multiple responses\n //If remote handler without response data, you can use rpc.call() for initiate broadcast calls.\n\n rpc.callBroadcast(\n 'getWorkerStat',\n { type: 'fullStat'}, //request parameters\n { //call options\n ttl: 1000, //wait response time (1 seconds), after run onComplete\n onResponse: function(err, stat) { //callback on each worker response\n all_stats[ stat.hostname+':'+ stat.pid ] = stat;\n\n },\n onComplete: function() { //callback on ttl expired\n console.log('----------------------- WORKER STATISTICS ----------------------------------------');\n for(var worker in all_stats) {\n s = all_stats[worker];\n console.log(worker, '\\tuptime=', s.uptime.toFixed(2) + ' seconds', '\\tcounter=', s.counter);\n }\n }\n });\n\n\nResults for three workers:\n\n ----------------------- WORKER STATISTICS ----------------------------------------\n host1:2612 \tuptime= 2470.39 seconds \tcounter= 2\n host2:1615 \tuptime= 3723.53 seconds \tcounter= 8\n host2:2822 \tuptime= 2279.16 seconds \tcounter= 3\n\nEugene Demchenko aka Goldy skype demchenkoe email demchenkoev@gmail.com\n", From 5b970969d80194fa45eae17fe6415261fc3ea663 Mon Sep 17 00:00:00 2001 From: Michele Rastelli Date: Mon, 7 Sep 2015 13:32:23 +0000 Subject: [PATCH 17/23] memory leak bugfix --- README.md | 10 ++++++---- example/round-robin/client.js | 25 +++++++++++++------------ example/round-robin/server.js | 2 +- index.js | 11 ++++++----- 4 files changed, 26 insertions(+), 22 deletions(-) diff --git a/README.md b/README.md index 9f6eafb..96d5f82 100644 --- a/README.md +++ b/README.md @@ -22,8 +22,9 @@ Run multiple servers.js for round-robin shared. ###server.js example var rpc = require('amqp-rpc').factory({ - url: "amqp://guest:guest@localhost:5672" - }); + exchange: "uuu-test", exchange_options: {exclusive: false, autoDelete: true}, + conn_options: {url: "amqp://guest:guest@rabbitmq:5672", heartbeat: 10} + }); rpc.on('inc', function(param, cb){ @@ -58,8 +59,9 @@ Run multiple servers.js for round-robin shared. ###client.js example var rpc = require('amqp-rpc').factory({ - url: "amqp://guest:guest@localhost:5672" - }); + exchange: "uuu-test", exchange_options: {exclusive: false, autoDelete: true}, + conn_options: {url: "amqp://guest:guest@rabbitmq:5672", heartbeat: 10} + }); rpc.call('inc', 5, function() { console.log('results of inc:', arguments); //output: [6,4,7] diff --git a/example/round-robin/client.js b/example/round-robin/client.js index 6d7640e..cdfc1c8 100644 --- a/example/round-robin/client.js +++ b/example/round-robin/client.js @@ -1,28 +1,29 @@ Object.defineProperty(Error, 'fromJSON', { - value: function (other) { - var err = new Error(); - Object.getOwnPropertyNames(other).forEach(function (key) { - err[key] = other[key]; - }); - return err; - }, - configurable: true + value: function (other) { + var err = new Error(); + Object.getOwnPropertyNames(other).forEach(function (key) { + err[key] = other[key]; + }); + return err; + }, + configurable: true }); var rpc = require('../../index').factory({ - conn_options: {url: "amqp://tdev1:mq4tdev1@127.0.0.1:5672", heartbeat: 10, confirm: true} + exchange: "uuu-test", exchange_options: {exclusive: false, autoDelete: true}, + conn_options: {url: "amqp://guest:guest@rabbitmq:5672", heartbeat: 10, confirm: true} }); rpc.rpcCall('inc', 5, null, null, function () { - console.log('results of inc:', arguments); //output: [6,4,7] + console.log('results of inc:', arguments); //output: [6,4,7] }); rpc.rpcCall('say.Hello', {name: 'John'}, null, null, function (msg) { - console.log('results of say.Hello:', msg); //output: Hello John! + console.log('results of say.Hello:', msg); //output: Hello John! }); rpc.rpcCall('withoutCB', {}, null, null, function (msg) { - console.log('withoutCB results:', msg); //output: please run function without cb parameter + console.log('withoutCB results:', msg); //output: please run function without cb parameter }); //rpc.rpcCall('withoutCB', {}); //output message on server side console diff --git a/example/round-robin/server.js b/example/round-robin/server.js index 1cba253..6eb3de9 100644 --- a/example/round-robin/server.js +++ b/example/round-robin/server.js @@ -14,7 +14,7 @@ Object.defineProperty(Error.prototype, 'toJSON', { var rpc = require('../../index').factory({ exchange: "uuu-test", exchange_options: {exclusive: false, autoDelete: true}, - conn_options: {url: "amqp://tdev1:mq4tdev1@127.0.0.1:5672", heartbeat: 10} + conn_options: {url: "amqp://guest:guest@rabbitmq:5672", heartbeat: 10} }); diff --git a/index.js b/index.js index bff79bf..949b09d 100644 --- a/index.js +++ b/index.js @@ -187,8 +187,8 @@ rpc.prototype.__onResult = function (message, headers, deliveryInfo) { cb.cb.apply(cb.context, args); - if (cb.autoDeleteCallback !== false) - delete this.__results_cb[deliveryInfo.correlationId]; + //if (cb.autoDeleteCallback !== false) + delete this.__results_cb[deliveryInfo.correlationId]; } /** @@ -219,8 +219,9 @@ rpc.prototype.rpcCall = function (cmd, params, options, context, cb) { $this.__results_cb[corr_id] = { cb: cb, - context: context, - autoDeleteCallback: !!options.autoDeleteCallback + context: context + //, + //autoDeleteCallback: !!options.autoDeleteCallback }; @@ -517,7 +518,7 @@ rpc.prototype.callBroadcast = function (cmd, params, options) { options || (options = {}); options.broadcast = true; - options.autoDeleteCallback = options.ttl ? false : true; + //options.autoDeleteCallback = options.ttl ? false : true; var corr_id = this.rpcCall.call(this, cmd, params, options, options.context, options.onResponse); if (options.ttl) { setTimeout(function () { From 267f389fd3fc678c88ab339a6e172f0fe398ca34 Mon Sep 17 00:00:00 2001 From: Michele Rastelli Date: Mon, 7 Sep 2015 13:37:51 +0000 Subject: [PATCH 18/23] memory leak bugfix --- README.md | 3 +++ package.json | 2 +- 2 files changed, 4 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index 96d5f82..41faef6 100644 --- a/README.md +++ b/README.md @@ -151,6 +151,9 @@ Eugene Demchenko aka Goldy skype demchenkoe email demchenkoev@gmail.com ## Changelog +0.2.4: + + * bugfix memory leak 0.2.3: diff --git a/package.json b/package.json index 6e741dc..b12418c 100644 --- a/package.json +++ b/package.json @@ -5,7 +5,7 @@ "amqp", "rpc" ], - "version": "0.2.3", + "version": "0.2.4", "preferGlobal": true, "author": { "name": "Eugene Demchenko" From f22b2d2d94d0d544aef1eeb66760e27541d041ba Mon Sep 17 00:00:00 2001 From: Michele Rastelli Date: Mon, 7 Sep 2015 14:40:26 +0000 Subject: [PATCH 19/23] memory leak bugfix --- index.js | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/index.js b/index.js index 949b09d..c899520 100644 --- a/index.js +++ b/index.js @@ -191,6 +191,16 @@ rpc.prototype.__onResult = function (message, headers, deliveryInfo) { delete this.__results_cb[deliveryInfo.correlationId]; } +rpc.prototype.setMessageTimeout = function (corr_id, timeout) { + var $this = this; + setTimeout(function () { + //release cb + if ($this.__results_cb[corr_id]) { + delete $this.__results_cb[corr_id]; + } + }, timeout); +}; + /** * call a remote command * @param {string} cmd command name @@ -206,8 +216,10 @@ rpc.prototype.rpcCall = function (cmd, params, options, context, cb) { if (!options) options = {}; + options.expiration = options.expiration || "20000"; options.contentType = 'application/json'; var corr_id = options.correlationId || uuid(); + $this.setMessageTimeout(corr_id, options.expiration); this._connect(function () { From 4c0afa0c9941aa3f494390e8e02eae66fac9ba33 Mon Sep 17 00:00:00 2001 From: Michele Rastelli Date: Wed, 4 Nov 2015 12:26:02 +0000 Subject: [PATCH 20/23] log info about amqp monitoring --- index.js | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/index.js b/index.js index c899520..e3da644 100644 --- a/index.js +++ b/index.js @@ -292,10 +292,20 @@ rpc.prototype._deleteBindings = function (queue, bindingsToDelete) { var self = this; bindingsToDelete.forEach(function (b) { queue.bind(b.exchange, b.routing, function () { + + if (bindingsToDelete.length > 0) { + console.log("======= Amqp monitoring: deleting preexisting bindings ======="); + } + bindingsToDelete.forEach(function (b) { console.log("deleting binding: ", b); queue.unbind(b.exchange, b.routing); }); + + if (bindingsToDelete.length > 0) { + console.log("=============================================================="); + } + }); }); }; @@ -547,10 +557,16 @@ rpc.prototype.printChannels = function (queueName) { var thisRabbit = this; thisRabbit._getChannels(queueName, function (channels) { if (channels.length > 0) { + + console.log("======= Amqp monitoring report ======="); + console.log("Channels in [" + queueName + "] queue: "); channels.forEach(function (v) { console.log("\t" + v); }); + + console.log("======================================"); + } }); }; From dea12bf8fd6962b3fb0b5d9740ab8845cab29d49 Mon Sep 17 00:00:00 2001 From: Michele Rastelli Date: Fri, 6 Nov 2015 16:20:23 +0000 Subject: [PATCH 21/23] bugfix discinnection problem --- index.js | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/index.js b/index.js index e3da644..91b266e 100644 --- a/index.js +++ b/index.js @@ -3,7 +3,10 @@ var uuid = require('node-uuid').v4; var os = require('os'); var queueNo = 0; +var instanceId = 0; + function rpc(opt) { + this.instanceId = instanceId++; if (!opt) opt = {}; this.opt = opt; @@ -86,9 +89,14 @@ rpc.prototype._connect = function (cb) { rpc.prototype.disconnect = function () { if (!this.__conn) return; - this.__conn.end(); + this.__conn.removeAllListeners('error'); + var self = this; + this.__conn.on('error', function () { + console.log('--> error disconnecting amqp-rpc #' + self.instanceId); + }); + this.__conn.disconnect(); this.__conn = null; -} +}; rpc.prototype._makeExchange = function (cb) { From d695c27f0b585e5c0ef76739c74481e20aff6625 Mon Sep 17 00:00:00 2001 From: Michele Rastelli Date: Fri, 4 Dec 2015 15:08:46 +0000 Subject: [PATCH 22/23] on onParallel renamed --- index.js | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/index.js b/index.js index 91b266e..1b4665a 100644 --- a/index.js +++ b/index.js @@ -341,7 +341,7 @@ rpc.prototype._deleteBindings = function (queue, bindingsToDelete) { */ -rpc.prototype.onParallel = function (cmd, cb, options, context) { +rpc.prototype.subscribeParallel = function (cmd, cb, options, context) { if (this.__cmds[cmd]) return false; options || (options = {}); @@ -412,7 +412,7 @@ rpc.prototype.onParallel = function (cmd, cb, options, context) { * @param options * @returns {boolean} */ -rpc.prototype.on = function (cmd, cb, options, context) { +rpc.prototype.subscribe = function (cmd, cb, options, context) { if (this.__cmds[cmd]) return false; options || (options = {}); From 1f2405e7bc24b935f61c8040985edbfd6c5d3819 Mon Sep 17 00:00:00 2001 From: Michele Rastelli Date: Fri, 4 Dec 2015 15:10:26 +0000 Subject: [PATCH 23/23] on onParallel renamed --- example/round-robin/server.js | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/example/round-robin/server.js b/example/round-robin/server.js index 6eb3de9..d50c5ea 100644 --- a/example/round-robin/server.js +++ b/example/round-robin/server.js @@ -18,13 +18,13 @@ var rpc = require('../../index').factory({ }); -rpc.on('zzttrr', function(param, cb){ +rpc.subscribe('zzttrr', function(param, cb){ var prevVal = param; var nextVal = param + 2; cb(++param, prevVal, nextVal); }, {queueName: "test_inc"}); -rpc.on('say.*', function (param, cb, inf) { +rpc.subscribe('say.*', function (param, cb, inf) { var arr = inf.cmd.split('.'); var name = (param && param.name) ? param.name : 'world'; @@ -33,7 +33,7 @@ rpc.on('say.*', function (param, cb, inf) { }); -rpc.on('withoutCB', function (param, cb, inf) { +rpc.subscribe('withoutCB', function (param, cb, inf) { if (cb) { cb('please run function without cb parameter') @@ -44,11 +44,11 @@ rpc.on('withoutCB', function (param, cb, inf) { }); -rpc.on('errorFn', function (param, cb) { +rpc.subscribe('errorFn', function (param, cb) { cb(new Error("errorFn"), null); }); -rpc.on('waitsTooMuch', function (param, cb) { +rpc.subscribe('waitsTooMuch', function (param, cb) { console.log("waitsTooMuch"); //cb("waitsTooMuch OK!"); setTimeout(cb.bind(null, "waitsTooMuch OK!"), 5000);