jakiro-comet-server

jakiro.js

/*
 * Jakiro
 * @desc   Yet Another Comet Server
 * @author Dongxu Huang <huangdx@rd.netease.com>
 * @date   2011-11-18
 */

var port = 8899 // default listen port
var createServer = require("http").createServer // createserver function
var puts  = require("util").puts // output log
var url = require("url")

var timeout = 50000
var buf_size = 10

// out put debug string
function log(msg)
{
    var time = new Date().toString().split(' ')[4]
    puts('\033[32;49;1m['+time+']\033[39;49;0m '+msg)
}

// get current time stamp
function get_timestamp()
{
    return new Date().getTime()
}

// send response to client
function response(res, code, body, callback)
{
    var type;
    body = JSON.stringify(body);
    if (callback)
    {
        body = callback + '(' + body + ');';
        type = 'application/javascript';
    }
    else
    {
        type = 'text/json';
    }
    res.writeHeader(code, {
        "Content-Type": type,
        "Content-Length": body.length
    });
    res.write(body);
    res.end();
}

// message body
function Message(time, data)
{
    this.time = time
    this.data = data
}

// comet channel
function Channel()
{
    // store messages
    var messages = []
    // current time stamp
    var time = get_timestamp()
    // waiting client's callbacks
    var callbacks = []
    
    // push data into a channel
    this.put = function(data)
    {
        time = get_timestamp()
        var message = new Message(time, data)
        messages.push(message)
        
        // call waiting clients' callback function
        while(callbacks.length > 0)
            callbacks.shift().callback([message])
        // clean message buffer
        while(messages.length > buf_size)
            messages.shift()
        return time
    };
    
    this.get = function(time, callback)
    {
        var i, message, new_msgs = []
        for (i = 0;i < messages.length ; i++) 
        {
            message = messages[i]
            if (message.time > time)
                new_msgs.push(message)
        }
        if (new_msgs.length > 0)
        {
            callback(new_msgs)
            return 'unread'
        }
        else
        {
            callbacks.push({time: get_timestamp(), callback: callback})
            return 'waiting'
        }
    };
    
    // update client time; client fetch server time
    this.update = function()
    {
        return time
    }
    
    // remove expire waiter or entire channel
    this.remove = function(expire)
    {
        // if this channel has no waiter in queue meanwhile live a long time
        var isdeletechannel = callbacks.length === 0 && time < expire
        
        while (callbacks.length > 0 && callbacks[0].time < expire)
        {
            callbacks.shift().callback([]);
        }
        return isdeletechannel
    };   
}

var channels = []

// get channel by name, or create new channel
function get_channel(name)
{
    var channel = channels[name]
    if (channel) 
        return channel
    channels[name] = new Channel()
    return channels[name];
}

// check expire channels or clients
function check_channels()
{
    var expire = get_timestamp() - timeout;
    for (channel in channels)
    {
        if (channels[channel].remove(expire))
        {
            delete channels[channel]
        }
    }
}
// check timeout
setInterval(check_channels, 1000)

function on_client_arrive(req, res)
{
    log("New Client From: " + req.connection.remoteAddress + " Query: " + req.url)
    
    var q = url.parse(req.url, true).query || {method:'update', channel:'test'}
    var channel_name = q.channel  || 'test'
    var method = q.method || 'update'
    var channel = get_channel(channel_name)
    var callback = q.callback || false
    
    // handle get
    if (method == 'get')
    {
        time = q.time || 0
        channel.get(time, function(messages)
        {
            time = channel.update()
            response(res, 200, {time:time, messages:messages}, callback)
        })
    }
    // handle put
    else if (method == 'put')
    {
        message = q.message || null
        time = channel.put(message)
        response(res, 200, {time:time}, callback);
    }
    // sync client time
    else{
        time = channel.update();
        response(res, 200, {time:time}, callback);
    }
}

var server = createServer(on_client_arrive);
server.listen(port);

log("Jakiro Server v0.01 Start ...")

yet another comet server written in node.js