Processing Laravel jobs from node.

action.js

'use strict';
var redis = require('redis');
var LaravelJob = require('./index');

// Create laravel job instance
var job = new LaravelJob({});

// Set handlers on the job 
job.registerHandler("Jobinja\\Jobs\\ExampleJob", function(job, payload) {
	console.log(payload);
});

// Listen for jobs on a redis client
// whenewhere a job is fired 
// it will be dispatched to the equivalant handler
job.makeRedisConnectionListen(redis.createClient());

index.js

'use strict';
module.exports = require('./src/laravel_job');

laravel_job.js

'use strict';
var serializer = require('./serializer');

var in_array = function (needle, stack) {
	for(var i in stack) {
		if (stack[i] == needle) return true;
	}

	return false;
}

var laravelJob = function (config) {
	this.setConfig(config);
};

laravelJob.prototype.setConfig = function(config) {
	this.config = config;
	this.handlers = {};
};

laravelJob.prototype.getConfig = function(hard) {
	if (hard && this.config === undefined) {
		throw "Config cannot be undefined";
	}
	return this.config;
}

laravelJob.prototype.registerHandler = function (jobClass, handler) {
	this.handlers[jobClass] = handler;
	return this;
}

laravelJob.unserialize = function (serialized) {
	return serializer.unserialize(serialized);
}

laravelJob.prototype.makeRedisConnectionListen = function (connection) {
	var that = this;
	function wiatRecursive() {
		connection.blpop('queues:default', 0, function (name, res) {
			that.dispatch(res);
			wiatRecursive();
		});
	}
	wiatRecursive();
}

laravelJob.prototype.dispatch = function (jobMeta) {
	var serialized,
		job,
		blacklist = ['queue', 'delay', 'job'],
		key,
		current,
		currentData,
		handler,
		jsonJob = jobMeta[1] ? jobMeta[1] : false;

	if (!jsonJob) return;

	job = JSON.parse(jsonJob);

	// Serialized part of the job
	// @TODO events and email are stored like commands
	serialized = job.data.command;

	// Unserialized data
	var data = serializer.unserialize(serialized);

	handler = data.className;

	// If we dont have any handler return 
	if (!this.handlers.hasOwnProperty(handler)) {
		return;
	}

	var payload = {};
	var pData = data.data;
	for (var i in pData) {

		key = i;
		current = currentData = pData[i];

		// Set data if we have current data
		if (typeof current === 'object' && current) {
			if (current.hasOwnProperty('data')) {
				currentData = current.data;
			}
		}

		// Check that PHP property is protected or private
		var start = String('\u0000*\u0000');
		if (i.substr(0, start.length) == start) {
			key = i.split('');
			for (var i = 0; i < start.length; i++) {
				key[i] = '';
			};
			key = key.join('');
		}

		if (in_array(key, blacklist)) continue;

		payload[key] = currentData;
	}

	return this.handlers[handler](jobMeta, payload);
}

module.exports = laravelJob;

serializer.js

module.exports = {
  unserialize : function (data) {
    //  discuss at: http://phpjs.org/functions/unserialize/
    // original by: Arpad Ray (mailto:[email protected])
    // improved by: Pedro Tainha (http://www.pedrotainha.com)
    // improved by: Kevin van Zonneveld (http://kevin.vanzonneveld.net)
    // improved by: Kevin van Zonneveld (http://kevin.vanzonneveld.net)
    // improved by: Chris
    // improved by: James
    // improved by: Le Torbi
    // improved by: Eli Skeggs
    // bugfixed by: dptr1988
    // bugfixed by: Kevin van Zonneveld (http://kevin.vanzonneveld.net)
    // bugfixed by: Brett Zamir (http://brett-zamir.me)
    //  revised by: d3x
    //    input by: Brett Zamir (http://brett-zamir.me)
    //    input by: Martin (http://www.erlenwiese.de/)
    //    input by: kilops
    //    input by: Jaroslaw Czarniak
    //        note: We feel the main purpose of this function should be to ease the transport of data between php & js
    //        note: Aiming for PHP-compatibility, we have to translate objects to arrays
    //   example 1: unserialize('a:3:{i:0;s:5:"Kevin";i:1;s:3:"van";i:2;s:9:"Zonneveld";}');
    //   returns 1: ['Kevin', 'van', 'Zonneveld']
    //   example 2: unserialize('a:3:{s:9:"firstName";s:5:"Kevin";s:7:"midName";s:3:"van";s:7:"surName";s:9:"Zonneveld";}');
    //   returns 2: {firstName: 'Kevin', midName: 'van', surName: 'Zonneveld'}

    var that = this,
      utf8Overhead = function (chr) {
        // http://phpjs.org/functions/unserialize:571#comment_95906
        var code = chr.charCodeAt(0);
        if (code < 0x0080) {
          return 0;
        }
        if (code < 0x0800) {
          return 1;
        }
        return 2;
      };
    error = function (type, msg, filename, line) {
      throw new that.window[type](msg, filename, line);
    };
    read_until = function (data, offset, stopchr) {
      var i = 2,
        buf = [],
        chr = data.slice(offset, offset + 1);

      while (chr != stopchr) {
        if ((i + offset) > data.length) {
          error('Error', 'Invalid');
        }
        buf.push(chr);
        chr = data.slice(offset + (i - 1), offset + i);
        i += 1;
      }
      return [buf.length, buf.join('')];
    };
    read_chrs = function (data, offset, length) {
      var i, chr, buf;

      buf = [];
      for (i = 0; i < length; i++) {
        chr = data.slice(offset + (i - 1), offset + i);
        buf.push(chr);
        length -= utf8Overhead(chr);
      }
      return [buf.length, buf.join('')];
    };
    _unserialize = function (data, offset) {
      var dtype, dataoffset, keyandchrs, keys, contig,
        length, array, readdata, readData, ccount,
        stringlength, i, key, kprops, kchrs, vprops,
        vchrs, value, chrs = 0,
        typeconvert = function (x) {
          return x;
        };

      if (!offset) {
        offset = 0;
      }
      dtype = (data.slice(offset, offset + 1))
        .toLowerCase();

      dataoffset = offset + 2;

      switch (dtype) {
      case 'i':
        typeconvert = function (x) {
          return parseInt(x, 10);
        };
        readData = read_until(data, dataoffset, ';');
        chrs = readData[0];
        readdata = readData[1];
        dataoffset += chrs + 1;
        break;
      case 'b':
        typeconvert = function (x) {
          return parseInt(x, 10) !== 0;
        };
        readData = read_until(data, dataoffset, ';');
        chrs = readData[0];
        readdata = readData[1];
        dataoffset += chrs + 1;
        break;
      case 'd':
        typeconvert = function (x) {
          return parseFloat(x);
        };
        readData = read_until(data, dataoffset, ';');
        chrs = readData[0];
        readdata = readData[1];
        dataoffset += chrs + 1;
        break;
      case 'n':
        readdata = null;
        break;
      case 's':
        ccount = read_until(data, dataoffset, ':');
        chrs = ccount[0];
        stringlength = ccount[1];
        dataoffset += chrs + 2;

        readData = read_chrs(data, dataoffset + 1, parseInt(stringlength, 10));
        chrs = readData[0];
        readdata = readData[1];
        dataoffset += chrs + 2;
        if (chrs != parseInt(stringlength, 10) && chrs != readdata.length) {
          error('SyntaxError', 'String length mismatch');
        }
        break;
      case 'a':
        readdata = {};

        keyandchrs = read_until(data, dataoffset, ':');
        chrs = keyandchrs[0];
        keys = keyandchrs[1];
        dataoffset += chrs + 2;

        length = parseInt(keys, 10);
        contig = true;

        for (i = 0; i < length; i++) {
          kprops = _unserialize(data, dataoffset);
          kchrs = kprops[1];
          key = kprops[2];
          dataoffset += kchrs;

          vprops = _unserialize(data, dataoffset);
          vchrs = vprops[1];
          value = vprops[2];
          dataoffset += vchrs;

          if (key !== i)
            contig = false;

          readdata[key] = value;
        }

        if (contig) {
          array = new Array(length);
          for (i = 0; i < length; i++)
            array[i] = readdata[i];
          readdata = array;
        }

        dataoffset += 1;
        break;
      case 'c':
      case 'o':
        var classdata = '';
        readdata = {};

        // class name
        ccount = read_until(data, dataoffset, ':');
        chrs = ccount[0];
        stringlength = ccount[1];
        dataoffset += chrs + 2;

        readData = read_chrs(data, dataoffset + 1, parseInt(stringlength, 10));
        chrs = readData[0];
        readdata.className = readData[1];
        dataoffset += chrs + 2;
        if (chrs != parseInt(stringlength, 10) && chrs != readdata.className.length) {
          error('SyntaxError', 'String length mismatch');
        }

        // class data
        if (dtype === 'c') {
          // custom serialization format
          ccount = read_until(data, dataoffset, ':');
          chrs = ccount[0];
          stringlength = ccount[1];
          dataoffset += chrs + 2;

          readData = read_chrs(data, dataoffset + 1, parseInt(stringlength, 10));
          chrs = readData[0];
          classdata = readData[1];

          readdata.data = _unserialize(data, dataoffset)[2];

          dataoffset += chrs + 1;
          if (chrs != parseInt(stringlength, 10) && chrs != classdata.length) {
            error('SyntaxError', 'String length mismatch');
          }
        }
        else {
          // default serialization format
          readdata.data = {};

          keyandchrs = read_until(data, dataoffset, ':');
          chrs = keyandchrs[0];
          keys = keyandchrs[1];
          dataoffset += chrs + 2;

          for (i = 0; i < parseInt(keys, 10); i++) {
            kprops = _unserialize(data, dataoffset);
            kchrs = kprops[1];
            key = kprops[2];
            dataoffset += kchrs;

            vprops = _unserialize(data, dataoffset);
            vchrs = vprops[1];
            value = vprops[2];
            dataoffset += vchrs;

            readdata.data[key] = value;
          }

          dataoffset += 1;
        }

        break;
      default:
        error('SyntaxError', 'Unknown / Unhandled data type(s): ' + dtype);
        break;
      }
      return [dtype, dataoffset - offset, typeconvert(readdata)];
    };

    return _unserialize((data + ''), 0)[2];
  }
};