Merge pull request #2 from preact/feature/add_queueing

When you need more jobs than cores on the jobserver

Author: techdubb

example.js

@@ -1,10 +1,25 @@
-var spark_jobserver = require('./spark_jobserver');
-var jobserver = new spark_jobserver();
+var spark_jobserver = require('spark_jobserver');
+var config = {
+  host: "localhost:8090",
+  queue: {
+    key: "<aws key>",
+    secret: "<aws secret>",
+    region: "us-east-1",
+    name: "/838405463324/jobs-queued"
+  }
+};
+
+var jobserver = new spark_jobserver(config);
 
 jobserver.jobs.start('theapp', 'com.foo.blah.awesome', {context: 'awesome', sync: true}, '', function(err, data){
   console.log(data);
 });
 
 jobserver.jobs.result('5ab8f3f9-c7d1-42ec-87cb-344d0d0b5c5c', function(err, data){
   console.log(data);
+});
+
+var config = '{"stress": {"test": {"duration": "15"}}}';
+jobserver.jobs.queue('testjob', 'spark.jobserver.test', {context: 'awesome'}, config, function(err, data){
+  console.log(data);
 });

example_worker.js

@@ -0,0 +1,52 @@
+/**
+ * NOTE: run `npm i sqs-worker` before doing this example
+ */
+var spark_jobserver = require('spark_jobserver'),
+  SQSWorker = require('sqs-worker');
+
+var config = {
+  host: "localhost:8090",
+  queue: {
+    key: "<aws key>",
+    secret: "<aws secret>",
+    region: "us-east-1",
+    name: "/838405463324/jobs-queued"
+  }
+};
+
+var endpoint = 'https://sqs.' + config.queue.region + '.amazonaws.com';
+var opts = {
+  endpoint: endpoint,
+  region: config.queue.region,
+  url: endpoint + config.queue.name
+};
+
+var queue_worker = new SQSWorker(opts, worker);
+var jobserver = new spark_jobserver({host: config.host});
+
+function worker(message, done) {
+  try {
+    message = JSON.parse(message);
+    console.log(message);
+  } catch (err) {
+    throw err;
+  }
+
+  jobserver.jobs.start(message.appName, message.classPath, message.options, message.body, function(err, data){
+    console.log(data);
+
+    var success = false;
+    success = (data.status == 'STARTED') || (data.status == 'ERROR') || (data.status == 'VALIDATION FAILED');
+
+    if ((data.status == 'ERROR') || (data.status == 'VALIDATION FAILED')) {
+      // let the humans know
+    }
+
+    if (data.status == 'NO SLOTS AVAILABLE') {
+      // sleep and try again later?
+    }
+
+    done(null, success);
+  });
+
+}

index.js

@@ -1,12 +1,10 @@
-// ala https://github.com/chadsmith/node-namecheap/blob/master/namecheap.js
 var request = require('request'),
   qs = require('querystring'),
   util = require('util');
 
-var spark_jobserver = function(host) {
-  host = typeof host !== 'undefined' ? host : 'localhost:8090';
-
-  this.endpoint = host;
+var spark_jobserver = function(options) {
+  this.options = options;
+  this.endpoint = options.host || 'localhost:8090';
 };
 
 spark_jobserver.prototype = {
@@ -49,6 +47,39 @@ spark_jobserver.prototype = {
         for (var attr in options) { qs[attr] = options[attr]; }
         return instance.command('jobs', qs, body, callback, 'POST');
       },
+      queue: function(app_name, class_path, options, body, callback) {
+        var AWS = '';
+        try {
+          AWS = require('aws-sdk');
+        } catch (ex) {
+          throw new Error("aws-sdk module not installed.");
+        }
+
+        var queue_config = instance.options.queue || '';
+        var queue = '';
+        var queue_url = '';
+        if (queue_config !== '') {
+          var region = queue_config.region || 'us-east-1';
+          queue_url = 'https://sqs.' + region + '.amazonaws.com' + queue_config.name;
+          queue = new AWS.SQS(queue_config);
+        }
+
+        if (queue === '') {
+          throw new Error("Queue not configured.");
+        }
+
+        var message = {
+          appName: app_name,
+          classPath: class_path,
+          options: options,
+          body: body
+        };
+
+        queue.sendMessage({
+          QueueUrl: queue_url,
+          MessageBody: JSON.stringify(message)
+        }, callback);
+      },
       result: function(job_id, callback) {
         return instance.command('jobs/' + job_id, {}, '', callback);
       },

package.json

@@ -18,5 +18,8 @@
   "license": "ISC",
   "dependencies": {
     "request": "^2.66.0"
+  },
+  "optionalDependencies": {
+    "aws-sdk": "^2.2.24"
   }
 }