connect/disconnect at each flush, tested with 30K metric/second, fix #4

Author: Clivern

README.md

@@ -12,7 +12,9 @@ Clone `statsd` and our new backend
 $ git clone https://github.com/statsd/statsd.git
 $ git clone https://github.com/uptimedog/statsd-rabbitmq-backend.git statsd/backends/statsd-rabbitmq-backend
 
-$ cd statsd/backends/statsd-rabbitmq-backend
+$ cd statsd
+$ npm install
+$ cd backends/statsd-rabbitmq-backend
 $ npm install
 $ cd ../..
 ```
@@ -23,7 +25,8 @@ Create a config file `config.js` like the following to use rabbitmq backend. Mor
 {
   amqp: {
       connection: "amqp://guest:guest@localhost:5672",
-      queue: "metrics"
+      queue: "metrics",
+      durable: false
   },
 
   backends: [ "./backends/statsd-rabbitmq-backend" ]
@@ -57,7 +60,7 @@ var open = require('amqplib').connect("amqp://guest:guest@localhost:5672");
 open.then(function(conn) {
     return conn.createChannel();
 }).then(function(ch) {
-    return ch.assertQueue(q).then(function(ok) {
+    return ch.assertQueue(q, {durable: false}).then(function(ok) {
         return ch.consume(q, function(msg) {
             if (msg !== null) {
                 console.log(msg.content.toString());

statsd-rabbitmq-backend/index.js

@@ -10,9 +10,9 @@
  *
  *   amqp.connection: RabbitMQ Connection eg. amqp://guest:guest@localhost:5672
  *   amqp.queue: RabbitMQ Exchange eg. metrics
+ *   ampq.durable: Whether the queue will survive a broker restart
  *
  */
-var util = require('util');
 
 
 function RabbitmqBackend(startupTime, config, emitter) {
@@ -21,8 +21,6 @@ function RabbitmqBackend(startupTime, config, emitter) {
 	this.lastException = startupTime;
 	this.config = config;
 
-	this.connection = require('amqplib').connect(this.config.amqp.connection);
-
 	// attach
 	emitter.on('flush', function(timestamp, metrics) {
 		self.flush(timestamp, metrics);
@@ -73,21 +71,21 @@ RabbitmqBackend.prototype.flush = function(timestamp, metrics) {
 
 	var queue = this.config.amqp.queue;
 	var msg = JSON.stringify(metric);
+	var durable = this.config.amqp.durable;
 
-	console.log('Sending metrics ', msg);
+	console.log('Attempt to send metrics ', msg);
 
 	// Publish
-	this.connection.then(function(conn) {
-		if (this.ch) {
-			return this.ch;
-		}
-		this.ch = conn.createChannel();
-		return this.ch;
-	}).then(function(ch) {
-		return ch.assertQueue(queue).then(function(ok) {
-			console.log('Sent ', msg);
-			return ch.sendToQueue(queue, Buffer.from(msg));
-		});
+	require('amqplib').connect(this.config.amqp.connection).then(function(conn) {
+		return conn.createChannel().then(function(ch) {
+			var ok = ch.assertQueue(queue, {durable: durable});
+
+			return ok.then(function(_qok) {
+				ch.sendToQueue(queue, Buffer.from(msg));
+				console.log("Sent ", msg);
+				return ch.close();
+			});
+		}).finally(function() { conn.close(); });
 	}).catch(console.warn);
 };