dataflo.ws/xzstream.js

var stream = require('stream');

var crc32 = require('buffer-crc32');

var liner = new stream.Transform ({objectMode: true});

function XZFormat (chunk) {
	this.buf = new Buffer (chunk, "binary");
}

if (!Buffer.prototype.equals)
Buffer.prototype.equals = function (a) {
	if (!Buffer.isBuffer(a)) return undefined;
	if (a.length !== this.length) return false;

	for (var i = 0; i < a.length; i++) {
		if (a[i] !== this[i]) return false;
	}

	return true;
};

function encMultiInt (num) {
	var buf = new Buffer (9);
	if (num > Math.pow (2, 52)) // max int size in js without losing precision
		return null;

	var i = 0;

	while (num >= 0x80) {
		buf.writeUInt8 (i, num & 127);
		num >>= 7;
		i++;
	}

	buf.writeUInt8 (i, num);

	return buf.slice (0, i);
}

function decMultiInt (buf) {

	var i = 0;
	var num;
	var byte;

	while (i < buf.length) {
		byte = buf.readUInt8 (i);
		num |= byte << i * 7;
		if (byte | 127 === 0) {
			break;
		}
	}

	return num;
}

//+-+-+-+-+-+-+-+-+-+-+-+-+=======+=======+     +=======+=======+-+-+-+-+-+-+-+-+-+-+-+-+
//|     Stream Header     | Block | Block | ... | Block | Index |     Stream Footer     |
//+-+-+-+-+-+-+-+-+-+-+-+-+=======+=======+     +=======+=======+-+-+-+-+-+-+-+-+-+-+-+-+

// we need to read at least 12 bytes, 6 — for magic, 2 for flags and 4 for CRC

//+---+---+---+---+---+---+-------+------+--+--+--+--+
//|  Header Magic Bytes   | Stream Flags |   CRC32   |
//+---+---+---+---+---+---+-------+------+--+--+--+--+

XZFormat.prototype.headerMagic = new Buffer ("\xFD7zXZ\x00", "binary");

XZFormat.prototype.readHeader = function () {
	if (this.buf.length < 12) {
		return;
	}

	console.log (this.buf);

	var magicOk = this.headerMagic.equals (this.buf.slice (0, 6));
	if (!magicOk) {
		throw "wrong magic!";
	}

	var nullByte = this.buf.readUInt8(6);
	var flags    = this.buf.readUInt8(7);

	// The first byte of Stream Flags is always a null byte. In the
	// future, this byte may be used to indicate a new Stream version
	// or other Stream properties.

	// The second byte of Stream Flags is a bit field:

	// Bit(s)  Mask  Description
	// 0-3    0x0F  Type of Check (see Section 3.4):
	// 4-7    0xF0  Reserved for future use; MUST be zero for now.
	var streamHasCRC32  = flags & 0x1 ? true: false;
	var streamHasCRC64  = flags & 0x4 ? true: false;
	var streamHasSHA256 = flags & 0xA ? true: false;

	var flagsB = new Buffer(2);
	flagsB.writeUInt8 (nullByte, 0);
	flagsB.writeUInt8 (flags, 1);
	var flagsCRC = crc32 (flagsB);

	var headerCRC = this.buf.slice (8, 12);

	// TODO: throw if reserved bits is set
	console.log (magicOk, flags, {
		CRC32:  streamHasCRC32,
		CRC64:  streamHasCRC64,
		SHA256: streamHasSHA256,
	}, flagsCRC, headerCRC);

	this.buf = this.buf.slice (12);

	this.streamHeaderRead = true;
}

//+-+-+-+-+---+---+---+---+-------+------+----------+---------+
//| CRC32 | Backward Size | Stream Flags | Footer Magic Bytes |
//+-+-+-+-+---+---+---+---+-------+------+----------+---------+

XZFormat.prototype.footerMagic = new Buffer ("YZ", "binary");

XZFormat.prototype.readFooter = function () {

}

//+==============+=================+===============+=======+
//| Block Header | Compressed Data | Block Padding | Check |
//+==============+=================+===============+=======+

function XZBlock (buf) {
	this.buf = buf;
}

//+-------------------+-------------+#################+
//| Block Header Size | Block Flags | Compressed Size |
//+-------------------+-------------+#################+

//---> +###################+======================+
//---> | Uncompressed Size | List of Filter Flags |
//---> +###################+======================+

//---> +================+--+--+--+--+
//---> | Header Padding |   CRC32   |
//---> +================+--+--+--+--+

XZBlock.prototype.read = function () {
	if (!this.headerRead && this.buf.length >= 20) {
		this.readHeader ()
		return;
	}
}

XZBlock.prototype.readHeader = function () {
	var blockHeaderSizeField = this.buf.readUInt8 (0)
	var blockHeaderSize = (blockHeaderSizeField + 1) * 4;
	// Bit(s)  Mask  Description
	// 0-1    0x03  Number of filters (1-4)
	// 2-5    0x3C  Reserved for future use; MUST be zero for now.
	// 6     0x40  The Compressed Size field is present.
	// 7     0x80  The Uncompressed Size field is present.
	var blockFlags      = this.buf.readUInt8 (1);
	var blockFilters    = blockFlags & 0x03;
	var blockReserved   = blockFlags & 0x3c;
	var haveComprSize   = blockFlags & 0x40;
	var haveUncomprSize = blockFlags & 0x80;
	var uncomprOffset = 0;

	var blockHeaderCRC = this.buf.slice (blockHeaderSize - 4, blockHeaderSize);

	if (blockReserved) {
		throw "reserved must be 0";
	}

	console.log (this.buf, blockHeaderSize, blockFlags, blockReserved, haveComprSize, haveUncomprSize);

	//if (haveComprSize) {
		// at most 9 bytes
		var comprSizeBuf = this.buf.slice (2, 11);
		console.log (comprSizeBuf);
		var comprSize    = decMultiInt (comprSizeBuf);
		uncomprOffset    = comprSizeBuf.length;
	//}

	console.log (uncomprOffset);

	//if (haveUncomprSize) {
		var uncomprSizeBuf = this.buf.slice (2 + uncomprOffset, 11 + uncomprOffset);
		var uncomprSize    = decMultiInt (uncomprSizeBuf);
	//}

	if (blockHeaderSizeField === 0) {
		// TODO: handle index
		throw "index not supported";
	}

	if (blockFilters) {
		throw "filters not supported";
	}

	// TODO: header padding bytes must be 0x00, if not, throw error

	console.log (
		"block header size:", blockHeaderSize,
		"compressed size:", comprSize,
		"uncompressed size:", uncomprSize,
		"header crc:", blockHeaderCRC
	);

}


XZFormat.prototype.append = function (chunk) {
	this.buf += chunk;
}

XZFormat.prototype.readBlock = function () {
	this.currentBlock = new XZBlock (this.buf);

	this.currentBlock.read();
}

liner._transform = function (chunk, encoding, done) {

	if (!this.xz) {
		this.xz = new XZFormat (chunk);
	} else {
		this.xz.append (chunk);
	}

	if (!this.xz.streamHeaderRead) {
		this.xz.readHeader();
		this.xz.readBlock();
		return;
	} else {

	}

	var data = chunk.toString();
	if (this._lastLineData) data = this._lastLineData + data;

	var lines = data.split('\n');
	this._lastLineData = lines.splice(lines.length-1,1)[0];

	lines.forEach(this.push.bind(this));
	done();
}

liner._flush = function (done) {
	if (this._lastLineData) this.push(this._lastLineData);
	this._lastLineData = null;
	done();
}

module.exports = liner;