server.js

/*
 * Blode - a simple, powerful syslog-like event broadcast daemon
 * Copyright (C) 2010 Ben LeMasurier
 *
 * This program can be distributed under the terms of the GNU GPL.
 * See the file COPYING.
*/

/*
 * (syslog like) message severity
 * 0: emerge,  1: alert
 * 2: crit,    3: err
 * 4: warning, 5: notice
 * 6: info,    7: debug,  8: none
 */

DEBUG = false;
HOST = "127.0.0.1";

require("./lib/Math.uuid");

var cluster = require("cluster");
var threads = require("os").cpus().length;
var derp =   require("./lib/util"),
    io   =   require("socket.io"),
    ws   =   require("websocket-server"),
    net  =   require("net"),
    util =   require("util"),
    url  =   require("url"),
    http =   require("http"),
    dgram =  require("dgram"),
    event =  require("events"),
    emitter = new event.EventEmitter,
    log_buffer = { id: 0, severity: 'none', message:  '--MARK--' };

function Client(stream, type) {
  this.type   = type;
  this.stream = stream;
  this.subscription = [
    'emerge', 'alert', 'crit', 'err', 'warning',
    'notice', 'info', 'debug', 'none'
  ];
}

(function() {
  if(cluster.isMaster) {
    // fork workers.
    for(var i = 0; i < threads; i++) {
      cluster.fork();
    }
  } else {
    var blode = {};

    blode.config = require('./config');
    blode.http   = require('./http');
    blode.udp    = require('./udp');

    blode.clients = [];

    blode.udp.listen(HOST, blode.config.dgram.log_port, emitter);
    blode.http.listen(HOST, blode.config.http.log_port, emitter);

    blode.tcp = {};
    blode.tcp.broadcast = net.createServer(function(stream) {
      stream.setEncoding('utf8');
      stream.on("connect", function() {
        var buffer = '';
        var client = new Client(stream, 'tcp');
        blode.clients.push(client);

        stream.on('error', function() {
          blode.clients.remove(client);
          client.stream.end();
        });

        stream.on('end', function() {
          blode.clients.remove(client);
          client.stream.end();
        });

        stream.on('data', function(data) {
          try {
            var subscribe = JSON.parse(data);
            if(subscribe instanceof Array)
              client.subscription = subscribe;
          } catch(e) { }
        });

      });
    }).listen(blode.config.socket.broadcast_port, HOST);
    util.puts("Event socket broadcast daemon started at " + HOST + ":" + blode.config.socket.broadcast_port);

    blode.io = {};
    blode.io.broadcast = io.listen(blode.config.io.port);
    blode.io.broadcast.on('connection', function(socket) {
      var client = new Client(socket, 'io');
      blode.clients.push(client);

      socket.on("disconnect", function() {
        blode.clients.remove(client);
      });
    });
    util.puts("socket.io broadcast daemon started at " + HOST + ":" + blode.config.io.port);

    emitter.on("log", function(log) {
      log.id = Math.uuid();

      blode.io.broadcast.emit('message', JSON.stringify(log));

      blode.clients.forEach(function(client) {
        try {
          if(client.subscription.indexOf(log.severity) != -1)
            if(client.type === 'tcp')
              client.stream.write(JSON.stringify(log) + "\r\n");
        } catch(e) {
          socket_clients.remove(client);
        }
      });

    });

  }
})();