cluster-rpc.js/master.js

'use strict';

function getInstanceMethods(inst) {
  var instanceMethods = Object.keys(inst)
    .map(function (key) { return 'function' === typeof inst[key] ? key : null; })
    .filter(function (key) { return key; })
    ;

  var protoMethods = Object.keys(Object.getPrototypeOf(inst))
    .map(function (key) { return 'function' === typeof Object.getPrototypeOf(inst)[key] ? key : null; })
    .filter(function (key) { return key; })
    ;

  return instanceMethods.concat(protoMethods);
}

function getMethods(inst, keys) {
  if (!keys) {
    keys = getInstanceMethods(inst);
  }

  return keys.filter(function (key) {
    if ('function' === typeof inst[key]) {
      return true;
    }
  });
}

function setup(opts) {
  var crypto = require('crypto');
  var methods = getMethods(opts.instance, opts.methods);
  var token = crypto.randomBytes(16).toString('hex');
  var inst = opts.instance;
  var prefixes = require('./prefixes.js').create(opts); // uses opts.name

  opts.master = opts.master || require('./process/master').create(prefixes);

  opts.master.on('connection', function (w) {
    if (opts.debug) { console.log('[cluster-rpc] [master] worker connected'); }
    w.send({
      methods: methods
    , _token: token
    , type: prefixes.init
    });

    w.on('message', function (cmd) {
      if (0 !== (cmd.type||'').indexOf(prefixes.root)) {
        //console.log('debug w: got unknown message type');
        return;
      }

      if (token !== cmd._token) {
        //console.log('debug w: got bad token');
        return;
      }

      if (!Array.isArray(cmd.args)) {
        throw new Error("[Sanity Fail] 'args' should be array of arguments");
      }

      switch (cmd.type) {
        case prefixes.rpc:
          cmd.args.push(function callback() {
            // args is probably err, data in most cases
            var args = Array.prototype.slice.call(arguments);

            w.send({
              args: args
            , id: cmd.id
            //, this: this
            , _token: token
            , type: prefixes.result
            });
          });

          inst[cmd.func].apply(inst, cmd.args);
          break;

        default:
          throw new Error("cluster-ipc UNKNOWN TYPE");
          //break;
      }

    });
  });
}

module.exports.create = function (opts) {
  if (opts.debug) { console.log('[cluster-rpc] [master] create'); }
  var cluster = require('cluster');
  var PromiseA = opts.PromiseA || global.Promise || require('bluebird');
  var init = false;

  opts._promise = PromiseA.resolve(opts.instance);
  opts._promise.addWorker = function (w) {
    if (opts.debug) { console.log('[cluster-rpc] [master] addWorker wrapper'); }
    if (!init) {
      init = true;
      setup(opts);
    }
    return opts.master.addWorker(w);
  };

  if (false !== opts.addOnFork) {
    if (opts.debug) { console.log('[cluster-rpc] [master] -- will call addWorker on each fork'); }
    cluster.on('fork', opts._promise.addWorker);
  }

  return opts._promise;
};