import net.jpountz.lz4.LZ4Factory; //导入依赖的package包/类
private LZ4Compressor(String type, Integer compressionLevel)
{
    this.compressorType = type;
    this.compressionLevel = compressionLevel;
    final LZ4Factory lz4Factory = LZ4Factory.fastestInstance();
    switch (type)
    {
        case LZ4_HIGH_COMPRESSOR:
        {
            compressor = lz4Factory.highCompressor(compressionLevel);
            break;
        }
        case LZ4_FAST_COMPRESSOR:
        default:
        {
            compressor = lz4Factory.fastCompressor();
        }
    }

    decompressor = lz4Factory.fastDecompressor();
}
 

import net.jpountz.lz4.LZ4Factory; //导入依赖的package包/类
@Override
public byte[] uncompress(byte[] data) throws IOException {
	LZ4Factory factory = LZ4Factory.fastestInstance();
	ByteArrayOutputStream baos = new ByteArrayOutputStream();
	LZ4FastDecompressor decompresser = factory.fastDecompressor();
	LZ4BlockInputStream lzis = new LZ4BlockInputStream(new ByteArrayInputStream(data), decompresser);
	
	int count;
	byte[] buffer = new byte[2048];
	while ((count = lzis.read(buffer)) != -1) {
		baos.write(buffer, 0, count);
	}
	lzis.close();
	
	return baos.toByteArray();
}
 

import net.jpountz.lz4.LZ4Factory; //导入依赖的package包/类
public void readFieldsC(DataInput in) throws IOException {
	int decompressedLength = in.readInt();
	int compressedLength = in.readInt();
	byte[] compressed = new byte[compressedLength];
	in.readFully(compressed, 0, compressedLength);
	LZ4Factory factory = LZ4Factory.fastestInstance();
	LZ4FastDecompressor decompressor = factory.fastDecompressor();
	byte[] uncompressedByteArray = new byte[decompressedLength];
	decompressor.decompress(compressed, 0, uncompressedByteArray, 0, decompressedLength);		
	//byte[] uncompressedByteArray = WritableUtils.readCompressedByteArray(in);
	UnsafeByteArrayInputStream inStream = new UnsafeByteArrayInputStream(uncompressedByteArray);

	size = inStream.readInt();
	tables = new HashMap<String, Table>();
	for (int i = 0; i < size; i++)
	{
		String tableName = inStream.readUTF();
		Table table = new Table(null, null);
		table.readFields(inStream);
		tables.put(tableName, table);
	}
}
 

import net.jpountz.lz4.LZ4Factory; //导入依赖的package包/类
public void writeC(DataOutput out) throws IOException {
	UnsafeByteArrayOutputStream outStream = new UnsafeByteArrayOutputStream();
	outStream.writeInt(tables.entrySet().size());
	for (Map.Entry<String, Table> entry : tables.entrySet())
	{
		String tableName = entry.getKey();
		Table table = entry.getValue();
		outStream.writeUTF(tableName);
		table.write(outStream);
	}
	outStream.flush();
	//WritableUtils.writeCompressedByteArray(out, outStream.toByteArray());
	LZ4Factory factory = LZ4Factory.fastestInstance();
	LZ4Compressor compressor = factory.fastCompressor();
	byte[] outStreamByteArray = outStream.toByteArray();
	int decompressedLength = outStreamByteArray.length;
	int maxCompressedLength = compressor.maxCompressedLength(decompressedLength);
	byte[] compressed = new byte[maxCompressedLength];
	int compressedLength = compressor.compress(outStreamByteArray, 0, decompressedLength, compressed, 0, maxCompressedLength);
	out.writeInt(decompressedLength);
	out.writeInt(compressedLength);
	out.write(compressed, 0, compressedLength);
	outStream.close();
}
 

import net.jpountz.lz4.LZ4Factory; //导入依赖的package包/类
@Override
protected void onDeserialize(Deserializer buf) {
    super.onDeserialize(buf);
    int length = buf.getInt(null);
    int compressedLength = buf.getInt(null);
    Preconditions.checkState(length == data.length,
            "Size of serialized sketch does not match expected value. Expected %s, actual %s.", data.length, length);

    if (length == compressedLength) {
        deserializeDataArray(data, length, buf);
    } else {
        LZ4FastDecompressor c = LZ4Factory.safeInstance().fastDecompressor();
        byte[] compressedData = buf.getBytes(null, compressedLength);
        c.decompress(compressedData, data);
    }
}
 

import net.jpountz.lz4.LZ4Factory; //导入依赖的package包/类
private void print(Request request) {
    Values ret = request.returnValues();
    CompressionType type = CompressionType.valueOf(ret.get(0).asInt8());
    int uncompressedSize = ret.get(1).asInt32();
    byte [] blob = ret.get(2).asData();
    if (type == CompressionType.LZ4) {
        LZ4Factory factory = LZ4Factory.fastestInstance();
        LZ4FastDecompressor decompressor = factory.fastDecompressor();
        byte [] uncompressed = new byte [uncompressedSize];
        int compressedLength = decompressor.decompress(blob, 0, uncompressed, 0, uncompressedSize);
        if (compressedLength != blob.length) {
            throw new DeserializationException("LZ4 decompression failed. compressed size does not match. Expected " + blob.length + ". Got " + compressedLength);
        }
        blob = uncompressed;
    }
    Slime slime = BinaryFormat.decode(blob);
    try {
        new JsonFormat(true).encode(System.out, slime);
    } catch (IOException e) {
        throw new RuntimeException(e);
    }
}
 

import net.jpountz.lz4.LZ4Factory; //导入依赖的package包/类
/**
 * lz4 解壓縮
 *
 * @param value
 * @return
 */
public static byte[] unlz4(byte[] value) {
	byte[] result = new byte[0];
	try {
		LZ4Factory factory = LZ4Factory.fastestInstance();
		LZ4FastDecompressor decompressor = factory.fastDecompressor();
		//
		final int INTEGER_BYTES = 4;
		// 取原始長度
		int uncompressedLength = ((value[0] & 0xFF) << 24) | ((value[1] & 0xFF) << 16) | ((value[2] & 0xFF) << 8)
				| ((value[3] & 0xFF));
		result = new byte[uncompressedLength];
		int read = decompressor.decompress(value, INTEGER_BYTES, result, 0, uncompressedLength);
		if (read != (value.length - INTEGER_BYTES)) {
			result = new byte[0];
		}
	} catch (Exception ex) {
		ex.printStackTrace();
	}
	return result;
}
 

import net.jpountz.lz4.LZ4Factory; //导入依赖的package包/类
/**
 * lz4 壓縮
 *
 * @param value
 * @return
 */
public static byte[] lz4(byte[] value) {
	byte[] result = new byte[0];
	try {
		LZ4Factory factory = LZ4Factory.fastestInstance();
		LZ4Compressor compressor = factory.fastCompressor();
		//
		final int INTEGER_BYTES = 4;
		int length = value.length;
		int maxCompressedLength = compressor.maxCompressedLength(length);
		byte[] buff = new byte[INTEGER_BYTES + maxCompressedLength];
		// 用4個byte紀錄原始長度
		buff[0] = (byte) (length >>> 24);
		buff[1] = (byte) (length >>> 16);
		buff[2] = (byte) (length >>> 8);
		buff[3] = (byte) (length);

		// 壓縮後長度
		int written = compressor.compress(value, 0, length, buff, INTEGER_BYTES, maxCompressedLength);
		// 新長度=4+壓縮後長度
		int newLength = INTEGER_BYTES + written;
		result = new byte[newLength];
		System.arraycopy(buff, 0, result, 0, newLength);
	} catch (Exception ex) {
		ex.printStackTrace();
	}
	return result;
}
 

import net.jpountz.lz4.LZ4Factory; //导入依赖的package包/类
LZ4CompressionCodec(@Nonnegative final int bodyLength) {
	if (bodyLength < 1 || bodyLength > MAX_BODY_LENGTH) {
		throw new IllegalArgumentException("bodyLength < 1 || bodyLength > 0xFEEC: " + bodyLength);
	}
	final LZ4Factory factory = LZ4Factory.fastestInstance();
	this.compressor = factory.fastCompressor();
	this.decompressor = factory.fastDecompressor();

	this.bodyLength = bodyLength;
	this.headerLength = HEADER_LENGTH;
	this.compressedLength = compressor.maxCompressedLength(bodyLength);
	this.footerLength = FOOTER_LENGTH;
	this.frameLength = headerLength + compressedLength + footerLength;

	this.compressInput = new byte[this.bodyLength];
	this.compressOutput = new byte[compressedLength];
	this.decompressInput = new byte[compressedLength];
	this.decompressOutput = new byte[this.bodyLength];
}
 

import net.jpountz.lz4.LZ4Factory; //导入依赖的package包/类
public CompressedImage(BufferedImage img, CompressedType type) {
	this.width = img.getWidth();
	this.height = img.getHeight();
	this.type = type;

	switch (type) {
	case PNG:
		this.data = ImageUtils.toPNG(img);
		this.uncompressedLength = -1;
		break;

	case LZ4:
		byte[] uncompressed = getUncompressedBytes(img);
		this.uncompressedLength = uncompressed.length;
		this.data = LZ4Factory.unsafeInstance().fastCompressor().compress(uncompressed);
		break;

	default:
		throw new IllegalArgumentException();
	}

}
 

import net.jpountz.lz4.LZ4Factory; //导入依赖的package包/类
public Image get() {
	switch (type) {
	case PNG:
		return ImageUtils.fromPNG(data);

	case LZ4:
		LZ4FastDecompressor lz4Decompressor = LZ4Factory.unsafeInstance().fastDecompressor();
		byte[] uncompressed = new byte[uncompressedLength];
		lz4Decompressor.decompress(data, uncompressed);
		return uncompressImage(width, height, uncompressed);
		// byte[]decompressed = lz4Decompressor.d
		// case LZ4:
		// return LZ4Factory.unsafeInstance().fastCompressor().compress(getUncompressedBytes(img));
		//
	default:
		throw new IllegalArgumentException();
	}
}
 

import net.jpountz.lz4.LZ4Factory; //导入依赖的package包/类
@Override
public byte[] compress(byte[] data) throws IOException {
	LZ4Factory factory = LZ4Factory.fastestInstance();
	ByteArrayOutputStream byteOutput = new ByteArrayOutputStream();
	LZ4Compressor compressor = factory.fastCompressor();
	LZ4BlockOutputStream compressedOutput = new LZ4BlockOutputStream(byteOutput, 2048, compressor);
	compressedOutput.write(data);
	compressedOutput.close();
	
	return byteOutput.toByteArray();
}
 

import net.jpountz.lz4.LZ4Factory; //导入依赖的package包/类
/**
 * Create a new {@link OutputStream} that will compress data using the LZ4 algorithm.
 *
 * @param out The output stream to compress
 * @param blockSize Default: 4. The block size used during compression. 4=64kb, 5=256kb, 6=1mb, 7=4mb. All other
 *            values will generate an exception
 * @param blockChecksum Default: false. When true, a XXHash32 checksum is computed and appended to the stream for
 *            every block of data
 * @param useBrokenFlagDescriptorChecksum Default: false. When true, writes an incorrect FrameDescriptor checksum
 *            compatible with older kafka clients.
 * @throws IOException
 */
public KafkaLZ4BlockOutputStream(OutputStream out, int blockSize, boolean blockChecksum, boolean useBrokenFlagDescriptorChecksum) throws IOException {
    super(out);
    compressor = LZ4Factory.fastestInstance().fastCompressor();
    checksum = XXHashFactory.fastestInstance().hash32();
    this.useBrokenFlagDescriptorChecksum = useBrokenFlagDescriptorChecksum;
    bd = new BD(blockSize);
    flg = new FLG(blockChecksum);
    bufferOffset = 0;
    maxBlockSize = bd.getBlockMaximumSize();
    buffer = new byte[maxBlockSize];
    compressedBuffer = new byte[compressor.maxCompressedLength(maxBlockSize)];
    finished = false;
    writeHeader();
}
 

import net.jpountz.lz4.LZ4Factory; //导入依赖的package包/类
public static byte[] compress(byte srcBytes[]) throws IOException {
    LZ4Factory factory = LZ4Factory.fastestInstance();
    ByteArrayOutputStream byteOutput = new ByteArrayOutputStream();
    LZ4Compressor compressor = factory.fastCompressor();
    LZ4BlockOutputStream compressedOutput = new LZ4BlockOutputStream(
            byteOutput, 2048, compressor);
    compressedOutput.write(srcBytes);
    compressedOutput.close();
    return byteOutput.toByteArray();
}
 

import net.jpountz.lz4.LZ4Factory; //导入依赖的package包/类
public static byte[] uncompress(byte[] bytes) throws IOException {
    LZ4Factory factory = LZ4Factory.fastestInstance();
    ByteArrayOutputStream baos = new ByteArrayOutputStream();
    LZ4FastDecompressor decompresser = factory.fastDecompressor();

    LZ4BlockInputStream lzis = new LZ4BlockInputStream(
            new ByteArrayInputStream(bytes), decompresser);
    int count;
    byte[] buffer = new byte[2048 * 256];
    while ((count = lzis.read(buffer)) != -1) {
        baos.write(buffer, 0, count);
    }
    lzis.close();
    return baos.toByteArray();
}
 

import net.jpountz.lz4.LZ4Factory; //导入依赖的package包/类
public static byte[] uncompressCGSS(byte[] src) throws IOException {
    byte[] buf = new byte[4];
    System.arraycopy(src, 4, buf, 0, 4);
    int destL = getInt(buf, 0);


    byte[] source = new byte[src.length - 16];
    byte[] dest = new byte[destL];
    System.arraycopy(src, 16, source, 0, src.length - 16);

    LZ4Factory factory = LZ4Factory.fastestInstance();
    LZ4FastDecompressor decompresser = factory.fastDecompressor();
    decompresser.decompress(source, dest, destL);
    return dest;
}
 

import net.jpountz.lz4.LZ4Factory; //导入依赖的package包/类
/**
 * Create a new {@link InputStream} that will decompress data using the LZ4 algorithm.
 *
 * @param in The stream to decompress
 * @param ignoreFlagDescriptorChecksum for compatibility with old kafka clients, ignore incorrect HC byte
 * @throws IOException
 */
public KafkaLZ4BlockInputStream(InputStream in, boolean ignoreFlagDescriptorChecksum) throws IOException {
    super(in);
    decompressor = LZ4Factory.fastestInstance().safeDecompressor();
    checksum = XXHashFactory.fastestInstance().hash32();
    this.ignoreFlagDescriptorChecksum = ignoreFlagDescriptorChecksum;
    readHeader();
    maxBlockSize = bd.getBlockMaximumSize();
    buffer = new byte[maxBlockSize];
    compressedBuffer = new byte[maxBlockSize];
    bufferOffset = 0;
    bufferSize = 0;
    finished = false;
}
 

import net.jpountz.lz4.LZ4Factory; //导入依赖的package包/类
protected void encodeAndCompressBody(ByteBuffer buffer, int startPosition) {
    int startOfBody = buffer.position();
    encodeBody(buffer);
    setEncodedBody(buffer, startOfBody, buffer.position() - startOfBody);
    length = buffer.position() - startPosition;

    if (compressionLimit != 0 && length-4 > compressionLimit) {
        byte[] compressedBody;
        compressionType = CompressionType.LZ4;
        LZ4Factory factory = LZ4Factory.fastestInstance();
        LZ4Compressor compressor = factory.fastCompressor();
        compressedBody = compressor.compress(encodedBody);

        log.log(LogLevel.DEBUG, "Uncompressed size: " + encodedBody.length + ", Compressed size: " + compressedBody.length);
        if (compressedBody.length + 4 < encodedBody.length) {
            buffer.position(startPosition);
            buffer.putInt(compressedBody.length + startOfBody - startPosition + 4 - 4);  // +4 for compressed size
            buffer.putInt(getCompressedCode(compressionType));
            buffer.position(startOfBody);
            buffer.putInt(encodedBody.length);
            buffer.put(compressedBody);
            buffer.limit(buffer.position());
            return;
        }
    }
    buffer.putInt(startPosition, length - 4); // Encoded length 4 less than actual length
    buffer.limit(buffer.position());
}
 

import net.jpountz.lz4.LZ4Factory; //导入依赖的package包/类
/**
 * Create a new {@link InputStream} that will decompress data using the LZ4 algorithm.
 *
 * @param in The stream to decompress
 * @throws IOException
 */
public LZ4FrameInputStream(InputStream in) throws IOException {
  super(in);
  decompressor = LZ4Factory.fastestInstance().safeDecompressor();
  checksum = XXHashFactory.fastestInstance().hash32();
  nextFrameInfo();
}
 

import net.jpountz.lz4.LZ4Factory; //导入依赖的package包/类
/**
 * Create a new {@link OutputStream} that will compress data using the LZ4 algorithm.
 *
 * @param out                             The output stream to compress
 * @param blockSize                       Default: 4. The block size used during compression. 4=64kb, 5=256kb, 6=1mb, 7=4mb. All other
 *                                        values will generate an exception
 * @param blockChecksum                   Default: false. When true, a XXHash32 checksum is computed and appended to the stream for
 *                                        every block of data
 * @param useBrokenFlagDescriptorChecksum Default: false. When true, writes an incorrect FrameDescriptor checksum
 *                                        compatible with older kafka clients.
 * @throws IOException
 */
public KafkaLZ4BlockOutputStream( OutputStream out, int blockSize, boolean blockChecksum, boolean useBrokenFlagDescriptorChecksum ) throws IOException {
    super( out );
    compressor = LZ4Factory.fastestInstance().highCompressor( 17 );
    checksum = XXHashFactory.fastestInstance().hash32();
    this.useBrokenFlagDescriptorChecksum = useBrokenFlagDescriptorChecksum;
    bd = new BD( blockSize );
    flg = new FLG( blockChecksum );
    bufferOffset = 0;
    maxBlockSize = bd.getBlockMaximumSize();
    buffer = new byte[maxBlockSize];
    compressedBuffer = new byte[compressor.maxCompressedLength( maxBlockSize )];
    finished = false;
    writeHeader();
}