abraxas - npm Package Compare versions

Comparing version 0.1.0 to 0.2.0

client.js

@@ -51,3 +51,3 @@ "use strict";
 
-        packets.acceptSerialWithErro('JOB_CREATED', function (error,data) {
+        packets.acceptSerialWithError('JOB_CREATED', function (error,data) {
             if (error) return task.acceptError(error);
@@ -54,0 +54,0 @@ self.handleJobResult(task,func,trace,packets,data);

package.json

 {
   "name": "abraxas",
   "description": "A streaming gearman client / worker / server (as you choose)",
-  "keywords": ["gearman","streaming","client","worker"],
+  "keywords": [
+    "gearman",
+    "streaming",
+    "client",
+    "worker"
+  ],
   "author": "Rebecca Turner <[e-mail]> (http://re-becca.org)",
-  "version": "0.1.0",
+  "version": "0.2.0",
   "dependencies": {
@@ -8,0 +13,0 @@ "gearman-packet": "^0.2.1",

README.md

@@ -87,2 +87,3 @@ Abraxas
     client and worker payloads, unless otherwise specified.
+  * *maxJobs* (default: 1) -- The maximum number of jobs to handle at once.
   * *debug* -- If true, unknown or unexpected packets will be logged with
@@ -89,0 +90,0 @@ console.error.  You can achieve the same result by listening for the

task-client.js

@@ -10,8 +10,19 @@ "use strict";
     this.options = options;
-    this.callback = callback;
     var accept = new stream.PassThrough(options.accept);
     var transmit = new stream.PassThrough(options.transmit);
+
+    if (! options.encoding) options.encoding = this.options.defaultEncoding;
+    if (options.encoding == 'buffer') options.encoding = null;
+    if (options.encoding && (!options.accept || !options.accept.encoding)) {
+        accept.setEncoding(options.encoding);
+    }
+    if (options.encoding && (!options.transmit || !options.transmit.encoding)) {
+        transmit.setEncoding(options.encoding);
+    }
+
     Task.call(this,accept,transmit,options);
-    // If we were given a callback then we aren't readable.
-    if (callback) accept.end();
+    if (callback) {
+        this.pipe(concat(function(data) { callback(null,data) }));
+        this.on('error', callback);
+    }
     if (options.nobody) transmit.end();
@@ -24,20 +35,7 @@ }
 ClientTask.prototype.acceptError = function (error) {
-    if (this.callback) {
-        this.callback(error);
-    }
-    else {
-        this.emit('error', error);
-    }
+    this.emit('error', error);
 }
 
 ClientTask.prototype.acceptResult = function (result) {
-    var callback = this.callback;
-    if (callback && result.pipe) {
-        result.pipe(concat(function(data){ callback(null,data) }));
-        result.on('error', callback);
-    }
-    else if (callback) {
-        callback(null,result);
-    }
-    else if (result == null) {
+    if (result == null) {
         this._reader.end();
@@ -44,0 +42,0 @@ }

worker.js

@@ -16,2 +16,9 @@ "use strict";
 
+    this._activeJobs = {};
+    this._activeJobsCount = 0;
+
+    if (!this.options.maxJobs) {
+        this.options.maxJobs = 1;
+    }
+
     var self = this;
@@ -44,2 +51,16 @@ this.packets.acceptDefault('NO_JOB', function(data) {
 
+Worker.startWork = function (jobid) {
+    this._activeJobs[jobid] = true;
+    if (this.options.maxJobs > ++ this._activeJobsCount) {
+        this.askForWork();
+    }
+}
+
+Worker.endWork = function (jobid) {
+    delete this._activeJobs[jobid];
+    if (this.options.maxJobs > -- this._activeJobsCount) {
+        this.askForWork();
+    }
+}
+
 Worker.unregisterWorker = function (func) {
@@ -111,2 +132,4 @@ if (!this._workers[func]) {
 
+    this.startWork(jobid);
+
     var options = {jobid: jobid, uniqueid: job.args.uniqueid, client: this};
@@ -131,2 +154,3 @@ if (worker.options.encoding) options.encoding = worker.options.encoding;
         }
+        self.endWork(jobid);
     });
@@ -138,3 +162,3 @@
         self.socket.write(end, options.encoding);
-        self.askForWork();
+        self.endWork(jobid);
     });
@@ -141,0 +165,0 @@

New alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Fixed alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Improved metrics

Total package byte prevSize

88728

increased by27.98%

Number of package files

26

increased by8.33%

Lines of code

1237

increased by2.4%

Number of lines in readme file

468

increased by0.21%

No dependency changes