ZeroMQ and Node.js pub/sub emitter

zemitter.js

// ZeroMQ PUB/SUB + Multicast (PGM) enabled event emitter for node.js
// gleicon - 2011

var util = require("util");
var events = require("events");
var zeromq = require("../node_modules/zeromq"); // check the right path
var sys = require("sys");

function DistEventEmitter(name, remote_node) {
  events.EventEmitter.call(this);
  this.id = new Date().getTime() + "-" + process.pid; 
  this.pub_socket = zeromq.createSocket('publisher');
  this.sub_socket = zeromq.createSocket('subscriber');
	
  this.pub_socket.connect("epgm://vnic0;225.0.0.73:12345", function(e){ if (e) console.log(e); });	
  this.sub_socket.connect("epgm://vnic0;225.0.0.73:12345", function(e){ if (e) console.log(e); });
	
  this.sub_socket.subscribe("EVENTS");
  this.sub_socket.on("message", DistEventEmitter.prototype._dispatch);

}

util.inherits(DistEventEmitter, events.EventEmitter);

DistEventEmitter.prototype._emit_local = DistEventEmitter.prototype.emit 

DistEventEmitter.prototype._dispatch = function (data) {
  DistEventEmitter.call(this);
  try {
    d = JSON.parse(data.toString('utf8').replace('EVENTS ', ''));
    console.log('r: '+d.msg)
    if (d.id != this.id) DistEventEmitter.prototype._emit_local(d.event_name, d.msg);			
	console.log('rec: '+d.msg);
    } catch (e) { console.log(e); }	
}

DistEventEmitter.prototype._broadcast = function(w) {
  DistEventEmitter.call(this);
  pkt = JSON.stringify({ "id": this.id,
	                 "event_emitter": "data_ws", 
			 "event_name" : "data",
			 "msg": w });

  this.pub_socket.send("EVENTS " + pkt);
}

DistEventEmitter.prototype.emit = function(evt, data) {
  this._broadcast(data)
  return this._emit_local(evt, data);
}

// test
var de = new DistEventEmitter();
de.on("data", function(data) { console.log('Received data: "' + data + '"'); })
de.on("error", function(data) { console.log('Received data: "' + data + '"'); })

de.emit("data", "It works!")
de.emit("data", "It works 2")