celery-client - npm Package Compare versions

Comparing version 0.1.8 to 0.1.9

handlers/redisHandler.js

@@ -6,13 +6,2 @@ 'use strict';
 
-function once(fn, context) {
-    var result;
-    return function () {
-        if (fn) {
-            result = fn.apply(context || this, arguments);
-            fn = null;
-        }
-        return result;
-    };
-}
-
 function retryStrategy(options) {
@@ -39,2 +28,20 @@ if (options.error) {
   self.taskKeyPrefix = 'celery-task-meta-';
+
+  self.resultHandlerCallbackMap = {};
+
+  self._resultHandler = self._handler.duplicate();
+  self._resultHandler.on('pmessage', function(pattern, channel, message) {
+    var taskId = self.fromResultKey(channel);
+
+    var _callback = self.resultHandlerCallbackMap[taskId];
+    delete self.resultHandlerCallbackMap[taskId];
+
+    if ('function' === typeof _callback) {
+      if (message) message = self.parseResult(message);
+      _callback(null, message);
+    }
+  });
+
+  var resultHandlerKey = self.createResultKey('*');
+  self._resultHandler.psubscribe(resultHandlerKey);
 };
@@ -102,2 +109,10 @@
 
+RedisHandler.prototype.fromResultKey = function(resultKey) {
+  var self = this;
+
+  var taskId = resultKey.replace(self.taskKeyPrefix, '');
+
+  return taskId;
+};
+
 RedisHandler.prototype.parseResult = function(rawResult) {
@@ -151,20 +166,13 @@ var self = this;
 
-  var resultHandler = self._handler.duplicate();
+  self.resultHandlerCallbackMap[taskId] = callback;
 
-  var resultCallback = once(function(channel, result) {
-    resultHandler.unsubscribe();
-    resultHandler.quit();
+  setTimeout(function() {
+    var _callback = self.resultHandlerCallbackMap[taskId];
+    delete self.resultHandlerCallbackMap[taskId];
 
-    if (result) {
-      result = self.parseResult(result);
+    if ('function' === typeof _callback) {
+      _callback(null, {status: 'TIMEOUT'});
     }
 
-    return callback(null, result);
-  });
-
-  setTimeout(resultCallback, 3000);
-  resultHandler.on('message', resultCallback);
-
-  var key = self.createResultKey(taskId);
-  resultHandler.subscribe(key);
+  }, 3000);
 };
@@ -171,0 +179,0 @@

package.json

 {
   "name": "celery-client",
-  "version": "0.1.8",
+  "version": "0.1.9",
   "description": "A client for Celery",
@@ -5,0 +5,0 @@ "main": "celery-client.js",

No alert changes

Improved metrics

Total package byte prevSize

14217

increased by2.76%

Lines of code

272

increased by1.12%

No dependency changes