node-ipc/lib/socketServer.js

var net     = require('net'),
    fs      = require('fs'),
    dgram   = require('dgram'),
    eventParser = require('../lib/eventParser.js'),
    pubsub  = require('event-pubsub'),
    Message = require('js-message');

function emit(socket, type, data){
    this.log('dispatching event to socket'.debug, ' : ', type.data, data);

    var message=new Message;
    message.type=type;
    message.data=data;

    if(this.config.rawBuffer){
        message=new Buffer(type,this.encoding);
    }else{
        message=eventParser.format(message);
    }

    if(this.udp4 || this.udp6){

        if(!socket.address || !socket.port){
            this.log('Attempting to emit to a single UDP socket without supplying socket address or port. Redispatching event as broadcast to all connected sockets');
            this.broadcast(type,data);
            return;
        }

        this.server.write(
            message,
            socket
        );
        return;
    };

    socket.write(message);
};

function broadcast(type,data){
    this.log('broadcasting event to all known sockets listening to '.debug, this.path.variable,' : ', ((this.port)?this.port:''),  type, data);
    var message=new Message;
    message.type=type;
    message.data=data;

    if(this.config.rawBuffer){
        message=new Buffer(type,this.encoding);
    }else{
        message=eventParser.format(message);
    }

    if(this.udp4 || this.udp6){
        for(var i=1, count=this.sockets.length; i<count; i++){
            this.server.write(message,this.sockets[i]);
        }
    }else{
        for(var i=0, count=this.sockets.length; i<count; i++){
            this.sockets[i].write(message);
        }
    }
};

function init(path,config,log,port){
    var server={
        config          : config,
        path            : path,
        port            : port,
        udp4            : false,
        udp6            : false,
        log             : log,
        server          : false,
        sockets         : [],
        emit            : emit,
        broadcast       : broadcast,
        define          : {
            listen      : {
                'get.events.broadcasting'   : 'does not require any special paramaters',
                'get.events.listening'      : 'does not require any special paramaters'
            },
            broadcast   : {
                'events.broadcasting'   : 'data.events is a JSON object of event definitions by type '+config.id+' will broadcast on '+path,
                'events.listening'      : 'data.events is a JSON object of event definitions by type '+config.id+' is listening for on '+path
            }
        },
        onStart         : function(socket){
            this.trigger(
                'start',
                socket
            );
        },
        start           : function(){
            if(!this.path){
                server.log('Socket Server Path not specified, refusing to start'.warn);
                return;
            }

            fs.unlink(
                this.path,
                (
                    function(server){
                        return function () {
                            server.log('starting server on '.debug,server.path.variable,((server.port)?':'+server.port:'').variable);

                            if(!server.udp4 && !server.udp6){
                                server.server=net.createServer(
                                    serverCreated
                                );
                            }else{
                                function UDPWrite(message,socket){
                                    var data=new Buffer(message, server.config.encoding);
                                    server.server.send(
                                        data,
                                        0,
                                        data.length,
                                        socket.port,
                                        socket.address,
                                        function(err, bytes) {
                                            if(err){
                                                server.trigger(
                                                    'error',
                                                    function(err){
                                                        server.trigger('error',err);
                                                    }
                                                );
                                            }
                                        }
                                    );
                                }

                                server.server=dgram.createSocket(
                                    ((server.udp4)? 'udp4':'udp6')
                                );
                                server.server.write=UDPWrite;
                                server.server.on(
                                    'listening',
                                    function () {
                                        serverCreated(server.server)
                                    }
                                );
                            }

                            function serverCreated(socket) {
                                server.sockets.push(socket);

                                if(socket.setEncoding)
                                    socket.setEncoding(server.config.encoding);

                                server.log('## socket connection to server detected ##'.rainbow);
                                socket.on(
                                    'close',
                                    function(socket){
                                        server.trigger(
                                            'close',
                                            socket
                                        );
                                    }
                                );

                                socket.on(
                                    'error',
                                    function(err){
                                        server.trigger('error',err);
                                    }
                                );

                                socket.on(
                                    'data',
                                    function(data,UDPSocket){
                                        var sock=((server.udp4 || server.udp6)? UDPSocket : socket);
                                        if(server.config.rawBuffer){
                                            data=new Buffer(data,this.encoding);
                                            server.trigger(
                                                'data',
                                                data,
                                                sock
                                            );
                                            return;
                                        }

                                        if(!this.ipcBuffer)
                                            this.ipcBuffer='';

                                        data=(this.ipcBuffer+=data);

                                        if(data.slice(-1)!=eventParser.delimiter || data.indexOf(eventParser.delimiter) == -1){
                                            server.log('Implementing larger buffer for this socket message. You may want to consider smaller messages'.notice);
                                            return;
                                        }

                                        this.ipcBuffer='';

                                        data=eventParser.parse(data);

                                        while(data.length>0){
                                            var message=new Message;
                                            message.load(data.shift());

                                            server.log('received event of : '.debug,message.type.data,message.data);

                                            if(message.data.id)
                                                sock.id=message.data.id;

                                            server.trigger(
                                                message.type,
                                                message.data,
                                                sock
                                            );
                                        }
                                    }
                                );

                                socket.on(
                                    'message',
                                    function(msg,rinfo) {
                                        if (!rinfo)
                                            return;
                                        server.log('Received UDP message from '.debug, rinfo.address.variable, rinfo.port);
                                        var data;

                                        if(server.config.rawSocket){
                                            data=new Buffer(msg,this.encoding);
                                        }else{
                                            data=msg.toString();
                                        }
                                        socket.emit('data',data,rinfo);
                                    }
                                );

                                server.trigger(
                                    'connect',
                                    socket
                                );

                                if(server.config.rawBuffer){
                                    return;
                                }

                                server.trigger(
                                    'get.events.broadcasting',
                                    socket
                                );

                                server.trigger(
                                    'get.events.listening',
                                    socket
                                );
                            }

                            function started(socket){
                                server.onStart(socket)
                            }

                            if(!port){
                                server.log('starting server as'.debug, 'Unix Socket'.variable);
                                server.server.listen(
                                    server.path,
                                    started
                                );

                                server.server.maxConnections=server.maxConnections;
                                return;
                            }

                            if(!server.udp4 && !server.udp4){
                                server.log('starting server as'.debug, 'TCP'.variable);
                                server.server.listen(
                                    server.port,
                                    server.path,
                                    started
                                );
                                return;
                            }

                            server.log('starting server as'.debug,((server.udp4)? 'udp4':'udp6').variable);
                            server.server.bind(
                                server.port,
                                server.path
                            );

                            started(
                                {
                                    address : server.path,
                                    port    : server.port
                                }
                            );
                        }
                    }
                )(this)
            );
        }
    };

    new pubsub(server);

    server.on(
        'get.events.broadcasting',
        function(socket){
            server.emit(
                socket,
                'events.broadcasting',
                {
                    id      : server.config.id,
                    events  : server.define.broadcast
                }
            );
        }
    );

    server.on(
        'get.events.listening',
        function(socket){
            server.emit(
                socket,
                'events.listening',
                {
                    id      : server.config.id,
                    events  : server.define.listen,
                }
            );
        }
    )

    server.on(
        'close',
        function(){
            for(var i=0, count=server.sockets.length; i<count; i++){
                var socket=server.sockets[i];
                var destroyedSocketId=false;

                if(socket){
                    if(socket.readable)
                        continue;
                }

                if(socket.id)
                   destroyedSocketId=socket.id;

                server.log('socket disconnected'.notice,' '+destroyedSocketId.variable);

                if(socket)
                    socket.destroy();

                server.sockets.splice(i,1);

                server.trigger('socket.disconnected', socket, destroyedSocketId);

                return;
            }
        }
    );

    return server;
}

module.exports=init;