safe error counting, close #11

titarenko · Feb 16, 2017 · a2ac760 · a2ac760
1 parent 4c9bbe7
commit a2ac760
Show file tree

Hide file tree

Showing 3 changed files with 8 additions and 11 deletions.
diff --git a/modules/builder.js b/modules/builder.js
@@ -74,7 +74,7 @@ function createRetryBuffer (channel) {
 			if (--data.ttl) {
 				channel.sendToQueue('worque-buffer', new Buffer(JSON.stringify(data)), { persistent: true });
 			} else {
-				channel.sendToQueue(data.name, new Buffer(JSON.stringify(_.pick(data, ['context', 'content']))), { persistent: true });
+				channel.sendToQueue(data.name, new Buffer(JSON.stringify(_.pick(data, ['context', 'content', 'failures']))), { persistent: true });
 			}
 			channel.ack(message);
 		});

diff --git a/modules/queue.js b/modules/queue.js
@@ -47,7 +47,7 @@ Queue.prototype.bind = function (source, pattern, options) {
 };
 
 Queue.prototype.publish = function (content, context, options) {
-	var data = { context: context, content: content };
+	var data = { context: context, content: content, failures: 0 };
 	var rawContent = new Buffer(JSON.stringify(data));
 	return this._exec(function (channel) {
 		return channel.sendToQueue(this._name, rawContent, options || { persistent: true });
@@ -64,11 +64,11 @@ Queue.prototype.subscribe = function (handler, options) {
 			try {
 				data = JSON.parse(rawContent.toString());
 			} catch (e) {
-				self.emit('failure', { error: error });
+				self.emit('failure', { error: error, name: self._name });
 				channel.ack(message);
 			}
 			self.emit('task', _.extend(data, { name: self._name }));
-			return Promise.resolve().then(function () {
+			return Promise.try(function () {
 				return handler.call(data.context, data.content);
 			}).tap(function (result) {
 				self.emit('result', _.extend({ result: result }, data));
@@ -112,12 +112,9 @@ Queue.prototype.retry = function (getTtl) {
 	}
 
 	return this._exec(function (channel) {
-		var errorCounter = 0, bufferQueue = this._bufferQueue;
-		this.on('result', function (ev) {
-			errorCounter = 0;
-		});
+		var bufferQueue = this._bufferQueue;
 		this.on('failure', function (ev) {
-			var ttl = getTtl(errorCounter++);
+			var ttl = getTtl(ev.failures++);
 			if (!ttl) {
 				return;
 			}
@@ -133,4 +130,4 @@ Queue.prototype.retry = function (getTtl) {
 			};
 		};
 	}
-};
+};
diff --git a/package.json b/package.json
@@ -1,6 +1,6 @@
 {
   "name": "worque",
-  "version": "0.10.6",
+  "version": "0.11.0",
   "description": "AMQP-based work queue",
   "main": "index.js",
   "scripts": {