index.js

var EventEmitter = require('events').EventEmitter;
var util = require('util');
var buffers = require('buffers');

var argument_length = {};
argument_length.C = 1;
argument_length.S = 2;
argument_length.s = 2;
argument_length.L = 4;
argument_length.l = 4;
argument_length.x = 1;

function ReadFormatRequest(format, callback)
{
    this.format = format;
    this.current_arg = 0;
    this.data = [];
    this.callback = callback;
}

function ReadFixedRequest(length, callback)
{
    this.length = length;
    this.callback = callback;
    this.data = new Buffer(length);
    this.received_bytes = 0;
}

ReadFixedRequest.prototype.execute = function(bufferlist)
{
    // TODO: this is a brute force version
    // replace with Buffer.slice calls
    var to_receive = this.length - this.received_bytes;
    for(var i=0 ; i < to_receive; ++i)
    {
        if (bufferlist.length == 0)
            return false;
        this.data[this.received_bytes++] = bufferlist.getbyte();
    }
    this.callback(this.data);
    return true;
}

ReadFormatRequest.prototype.execute = function(bufferlist, tag1, tag2)
{
    while (this.current_arg < this.format.length)
    {
        var arg = this.format[this.current_arg];
        if (bufferlist.length < argument_length[arg])
            return false; // need to wait for more data to prcess this argument

        // TODO: measure Buffer.readIntXXX performance and use them if faster
        // note: 4 and 2-byte values may cross chunk border & split. need to handle this correctly
        // maybe best approach is to wait all data required for format and then process fixed buffer
        // TODO: byte order!!!
        switch (arg) {
        case 'C': 
            this.data.push(bufferlist.getbyte());
            break;
        case 'S':
        case 's':
            var b1 = bufferlist.getbyte();
            var b2 = bufferlist.getbyte();
            this.data.push(b2*256+b1);
            break;
        case 'l':
        case 'L':
            var b1 = bufferlist.getbyte();
            var b2 = bufferlist.getbyte();
            var b3 = bufferlist.getbyte();
            var b4 = bufferlist.getbyte();
            this.data.push(((b4*256+b3)*256 + b2)*256 + b1);
            break;
        case 'x':
            bufferlist.getbyte();
            break;
        }
        this.current_arg++;
    }
    this.callback(this.data);
    return true;
}

function UnpackStream()
{
    EventEmitter.call(this);

    this.readlist = [];
    this.length = 0;
    this.offset = 0;
    this.read_queue = [];
    this.write_queue = [];
    this.write_length = 0;
}
util.inherits(UnpackStream, EventEmitter);

UnpackStream.prototype.write = function(buf)
{
    this.readlist.push(buf);
    this.length += buf.length;
    this.resume();
}

UnpackStream.prototype.pipe = function(stream)
{
    // TODO: ondrain & pause
    this.on('data', function(data)
    {
        stream.write(data);
    });
}

UnpackStream.prototype.unpack = function(format, callback)
{
    this.read_queue.push(new ReadFormatRequest(format, callback));
    this.resume();
}

UnpackStream.prototype.unpackTo = function(destination, names_formats, callback)
{
    var names = [];
    var format = '';
    
    for (var i=0; i < names_formats.length; ++i)
    {
        var off = 0;
        while(off < names_formats[i].length && names_formats[i][off] == 'x')
        {
            format += 'x';
            off++;
        }

        if (off < names_formats[i].length)
        {
            format += names_formats[i][off];
            var name = names_formats[i].substr(off+2);
            names.push(name);
        }
    }

    this.unpack(format, function(data) {
        if (data.length != names.length)
            throw 'Number of arguments mismatch, ' + names.length + ' fields and ' + data.length + ' arguments';
        for (var fld = 0; fld < data.length; ++fld)
        {
            destination[names[fld]] = data[fld];
        }
        callback(destination);
    });
}

UnpackStream.prototype.get = function(length, callback)
{
    this.read_queue.push(new ReadFixedRequest(length, callback));
    this.resume();
}

UnpackStream.prototype.resume = function()
{
    if (this.resumed)
         return;
    this.resumed = true;
    // process all read requests until enough data in the buffer
    while(this.read_queue[0].execute(this))
    {
        this.read_queue.shift();
        if (this.read_queue.length == 0)
            return;
    }
    this.resumed = false;
}

UnpackStream.prototype.getbyte = function()
{   
    var res = 0;
    var b = this.readlist[0];
    if (this.offset + 1 < b.length)
    {
        res = b[this.offset];
        this.offset++;
        this.length--;
   
    } else {

        // last byte in current buffer, shift read list
        res = b[this.offset];
        this.readlist.shift();
        this.length--;
        this.offset = 0;
    }
    return res;
}

// TODO: measure node 0.5+ buffer serialisers performance
UnpackStream.prototype.pack = function(format, args)
{
    var packetlength = 0;
 
    var arg = 0;
    for (var i = 0; i < format.length; ++i)
    {
        var f = format[i];
        if (f == 'x')
        {
            packetlength++;
        } else if (f == 'p') {
            packetlength += xutil.padded_length(args[arg++].length);
        } else if (f == 'a') {
            packetlength += args[arg].length;
            arg++;
        } else {
            // this is a fixed-length format, get length from argument_length table
            packetlength += argument_length[f];
            arg++;
        }
    }

    var buf = new Buffer(packetlength);
    var offset = 0;
    var arg = 0;
    for (var i = 0; i < format.length; ++i)
    {
        switch(format[i])
        {
            case 'x': 
                buf[offset++] = 0;
                break;
            case 'C':
                var n = args[arg++];
                buf[offset++] = n;
                break;
            case 's': // TODO: implement signed INT16!!!
            case 'S':
                var n = args[arg++];
                buf[offset++] = n & 0xff;
                buf[offset++] = (n >> 8) & 0xff;
                break;
            case 'l': // TODO: implement signed INT32!!!
            case 'L':
                var n = args[arg++];
                buf[offset++] = n & 0xff;
                buf[offset++] = (n >> 8) & 0xff;
                buf[offset++] = (n >> 16) & 0xff;
                buf[offset++] = (n >> 24) & 0xff;
                break;
            case 'a':  // string or buffer
                var str = args[arg++];
                if (Buffer.isBuffer(str))
                {
                    str.copy(buf, offset);
                    offset += str.length;
                } else {
                    // TODO: buffer.write could be faster
                    for (var c = 0; c < str.length; ++c)
                        buf[offset++] = str.charCodeAt(c);
                }
                break;
            case 'p':  // padded string
                var str = args[arg++];
                var len = xutil.padded_length(str.length);
                // TODO: buffer.write could be faster
                var c = 0;
                for (; c < str.length; ++c)
                    buf[offset++] = str.charCodeAt(c);
                for (; c < len; ++c)
                    buf[offset++] = 0;
                break;
        }         
    }
    this.write_queue.push(buf);
    this.write_length += buf.length;
    return this;
}

UnpackStream.prototype.flush = function(stream)
{
    // TODO: measure performance benefit of 
    // creating and writing one big concatenated buffer

    // TODO: check write result
    // pause/resume streaming
    for (var i=0; i < this.write_queue.length; ++i)
    {
         //stream.write(this.write_queue[i])
         this.emit('data', this.write_queue[i]);
    }
    this.write_queue = [];
    this.write_length = 0;
}

module.exports = UnpackStream;