bus.js

// Generated by CoffeeScript 1.9.2
var Publish, Receive, async, createhub, cuid, mkdirp, resolve, shulz;

cuid = require('cuid');

async = require('odo-async');

Publish = require('./publish');

Receive = require('./receive');

shulz = require('shulz');

resolve = require('path').resolve;

mkdirp = require('mkdirp');

createhub = function() {
  return require('odo-hub/hub')(require('odo-hub/dispatch_parallel')());
};

module.exports = function(options) {
  var _, _fin, _ondrained, _onincomingempty, _onoutgoingempty, _removedestination, _subscribe, _unsubscribe, address, addresses, advertise, bind, datadir, envelope, exec, hub, i, incoming, len, outgoing, publisher, rawhub, receiver, ref, ref1, ref2, ref3, replayincoming, replayoutgoing, send, subscriptions;
  _fin = false;
  advertise = options.advertise;
  if (advertise == null) {
    advertise = options.bind;
  }
  bind = options.bind;
  datadir = resolve(process.cwd(), options.datadir);
  mkdirp.sync(datadir);
  incoming = options.incoming;
  if (incoming == null) {
    incoming = shulz.open(resolve(datadir, './incoming.shulz'));
  }
  outgoing = options.outgoing;
  if (outgoing == null) {
    outgoing = shulz.open(resolve(datadir, './outgoing.shulz'));
  }
  subscriptions = options.subscriptions;
  if (subscriptions == null) {
    subscriptions = shulz.open(resolve(datadir, './subscriptions.shulz'));
  }
  hub = options.hub;
  if (hub == null) {
    hub = createhub();
  }
  rawhub = options.rawhub;
  if (rawhub == null) {
    rawhub = createhub();
  }
  _ondrained = [];
  _onoutgoingempty = function() {
    var cb, i, len;
    if (incoming.length() !== 0) {
      return;
    }
    for (i = 0, len = _ondrained.length; i < len; i++) {
      cb = _ondrained[i];
      cb();
    }
    return _ondrained = [];
  };
  _onincomingempty = function() {
    var cb, i, len;
    if (outgoing.length() !== 0) {
      return;
    }
    for (i = 0, len = _ondrained.length; i < len; i++) {
      cb = _ondrained[i];
      cb();
    }
    return _ondrained = [];
  };
  publisher = Publish();
  exec = function(envelope, cb) {
    var fn, i, key, len, ref, tasks;
    tasks = [];
    ref = envelope.keys;
    fn = function(key) {
      return tasks.push(function(cb) {
        rawhub.emit(key, envelope, cb);
        return hub.emit(key, envelope.data, cb);
      });
    };
    for (i = 0, len = ref.length; i < len; i++) {
      key = ref[i];
      fn(key);
    }
    return async.parallel(tasks, cb);
  };
  replayincoming = [];
  ref = incoming.all();
  for (_ in ref) {
    envelope = ref[_];
    replayincoming.push(envelope);
  }
  async.delay(function() {
    var i, len, results;
    results = [];
    for (i = 0, len = replayincoming.length; i < len; i++) {
      envelope = replayincoming[i];
      results.push((function(envelope) {
        return exec(envelope, function() {
          incoming.clear(envelope.id);
          if (incoming.length() === 0) {
            return _onincomingempty();
          }
        });
      })(envelope));
    }
    return results;
  });
  receiver = Receive(bind, function(envelope, done) {
    envelope = JSON.parse(envelope.toString());
    incoming.set(envelope.id, envelope);
    async.delay(function() {
      return exec(envelope, function() {
        incoming.clear(envelope.id);
        if (incoming.length() === 0) {
          return _onincomingempty();
        }
      });
    });
    return done();
  });
  ref1 = subscriptions.all();
  for (_ in ref1) {
    addresses = ref1[_];
    for (address in addresses) {
      _ = addresses[address];
      publisher.register(address, address);
    }
  }
  _subscribe = function(key, address) {
    var subs;
    subs = subscriptions.get(key);
    if (subs == null) {
      subs = {};
    }
    subs[address] = true;
    subscriptions.set(key, subs);
    return publisher.register(address, address);
  };
  _unsubscribe = function(key, address) {
    var subs;
    subs = subscriptions.get(key);
    if (subs == null) {
      subs = {};
    }
    delete subs[address];
    return subscriptions.set(key, subs);
  };
  _removedestination = function(key, address) {
    var ref2, results, subs;
    ref2 = subscriptions.all();
    results = [];
    for (key in ref2) {
      subs = ref2[key];
      if (subs[address] == null) {
        continue;
      }
      delete subs[address];
      results.push(subscriptions.set(key, subs));
    }
    return results;
  };
  hub.every('_subscribe', function(m, cb) {
    var i, key, len, ref2;
    ref2 = m.keys;
    for (i = 0, len = ref2.length; i < len; i++) {
      key = ref2[i];
      _subscribe(key, m.address);
    }
    return cb();
  });
  hub.every('_unsubscribe', function(m, cb) {
    var i, key, len, ref2;
    ref2 = m.keys;
    for (i = 0, len = ref2.length; i < len; i++) {
      key = ref2[i];
      _unsubscribe(key, m.address);
    }
    return cb();
  });
  hub.every('_removedestination', function(m, cb) {
    _removedestination(m.address);
    return cb();
  });
  replayoutgoing = [];
  ref2 = outgoing.all();
  for (_ in ref2) {
    envelope = ref2[_];
    replayoutgoing.push(envelope);
    ref3 = envelope.addresses;
    for (i = 0, len = ref3.length; i < len; i++) {
      address = ref3[i];
      publisher.register(address, address);
    }
  }
  async.delay(function() {
    var j, len1, results;
    results = [];
    for (j = 0, len1 = replayoutgoing.length; j < len1; j++) {
      envelope = replayoutgoing[j];
      results.push((function(envelope) {
        var message;
        message = JSON.stringify(envelope);
        return publisher.publish(envelope.id, message, envelope.addresses, function() {
          outgoing.clear(envelope.id);
          if (outgoing.length() === 0) {
            return _onoutgoingempty();
          }
        });
      })(envelope));
    }
    return results;
  });
  send = function(addresses, msgid, keys, data, cb) {
    var message;
    envelope = {
      id: msgid,
      keys: keys,
      addresses: addresses,
      sent: new Date(),
      data: data
    };
    outgoing.set(msgid, envelope);
    message = JSON.stringify(envelope);
    return publisher.publish(msgid, message, addresses, function() {
      outgoing.clear(msgid);
      if (outgoing.length() === 0) {
        _onoutgoingempty();
      }
      if (cb != null) {
        return async.delay(cb);
      }
    });
  };
  return {
    publish: function(keys, data) {
      var j, key, len1, msgid, subs;
      if (!(keys instanceof Array)) {
        keys = [keys];
      }
      addresses = {};
      for (j = 0, len1 = keys.length; j < len1; j++) {
        key = keys[j];
        subs = subscriptions.get(key);
        if (subs == null) {
          continue;
        }
        for (address in subs) {
          _ = subs[address];
          addresses[address] = true;
        }
      }
      addresses = Object.keys(addresses);
      msgid = cuid();
      if (addresses.length === 0) {
        return msgid;
      }
      send(addresses, msgid, keys, data);
      return msgid;
    },
    send: function(key, data, address) {
      var msgid;
      publisher.register(address, address);
      msgid = cuid();
      send([address], msgid, [key], data);
      return msgid;
    },
    subscribe: function(address, keys) {
      var fn, j, key, len1, subscribemsgids;
      if (!(keys instanceof Array)) {
        keys = [keys];
      }
      subscribemsgids = [];
      publisher.register(address, address);
      fn = function(key) {
        var data, msgid;
        msgid = cuid();
        subscribemsgids.push(msgid);
        data = {
          keys: keys,
          address: advertise
        };
        return send(address, msgid, ['_subscribe'], data);
      };
      for (j = 0, len1 = keys.length; j < len1; j++) {
        key = keys[j];
        fn(key);
      }
      return subscribemsgids;
    },
    unsubscribe: function(address, keys) {
      var fn, j, key, len1, unsubscribemsgids;
      if (!(keys instanceof Array)) {
        keys = [keys];
      }
      unsubscribemsgids = [];
      publisher.register(address, address);
      fn = function(key) {
        var data, msgid;
        msgid = cuid();
        unsubscribemsgids.push(msgid);
        data = {
          keys: keys,
          address: advertise
        };
        return send(address, msgid, ['_unsubscribe'], data);
      };
      for (j = 0, len1 = keys.length; j < len1; j++) {
        key = keys[j];
        fn(key);
      }
      return unsubscribemsgids;
    },
    removedestination: function(address) {
      var data, msgid;
      publisher.register(address, address);
      msgid = cuid();
      data = {
        address: advertise
      };
      send(address, msgid, ['_removedestination'], data);
      return msgid;
    },
    drain: function(cb) {
      if (incoming.length() === 0 && outgoing.length() === 0) {
        async.delay(cb);
        return;
      }
      return _ondrained.push(cb);
    },
    status: function() {
      return (subscriptions.length()) + " subscribers, " + (incoming.length()) + " incoming and " + (outgoing.length()) + " outgoing";
    },
    close: function() {
      if (_fin) {
        return;
      }
      _fin = true;
      publisher.close();
      receiver.close();
      subscriptions.close();
      incoming.close();
      return outgoing.close();
    },
    every: hub.every,
    once: hub.once,
    any: hub.any,
    all: hub.all,
    hub: hub,
    raw: rawhub
  };
};