package com.yiidata.intergration.common.cache.serde;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import com.esotericsoftware.kryo.util.Pool;
import com.google.common.primitives.Ints;
import com.google.gson.Gson;
import org.apache.commons.lang.StringUtils;

import java.io.ByteArrayInputStream;
import java.nio.charset.Charset;
import java.util.Arrays;


/**
 *
 * Kryo 序列化
 *
 * <pre>
 *
 * Created by zhenqin.
 * User: zhenqin
 * Date: 2022/5/30
 * Time: 下午4:41
 * Vendor: yiidata.com
 *
 * </pre>
 *
 * @author zhenqin
 */
public class KryoFixSerializer<T> implements Serializer<T> {

    // Pool constructor arguments: thread safe, soft references, maximum capacity
    final Pool<Kryo> kryoPool = new Pool<Kryo>(true, false, 8) {
        protected Kryo create () {
            Kryo kryo = new Kryo();
            // Kryo 配置
            return kryo;
        }
    };



    private final StringSerializer serializer;

    public KryoFixSerializer() {
        serializer = new StringSerializer();
    }


    public KryoFixSerializer(String charset) {
        this.serializer = new StringSerializer(Charset.forName(charset));
    }


    public KryoFixSerializer(StringSerializer serializer) {
        this.serializer = serializer;
    }

    @Override
    public byte[] serialize(T o) {
        // 获取池中的Kryo对象
        Kryo kryo = kryoPool.obtain();
        try {
            String clazz = o.getClass().getName();
            byte[] classString = clazz.getBytes(serializer.getCharset());

            int length = classString.length;
            byte[] head = Ints.toByteArray(length);

            // 写入数据缓冲区
            Output opt = new Output(1024, -1);
            kryo.writeClassAndObject(opt, o);
            opt.flush();
            byte[] body = opt.getBuffer();

            byte[] bytes = new byte[head.length + length + body.length];
            System.arraycopy(head, 0, bytes, 0, head.length);
            System.arraycopy(classString, 0, bytes, head.length, classString.length);
            System.arraycopy(body, 0, bytes, head.length + length, body.length);
            return bytes;
        } finally {
            // 将kryo对象归还到池中
            kryoPool.free(kryo);
        }
    }

    @Override
    public T deserialize(byte[] bytes) {
        // 获取池中的Kryo对象
        Kryo kryo = kryoPool.obtain();
        try {
            // 前面 4 个字节是头，代表 类名 的长度
            byte[] head = new byte[4];
            System.arraycopy(bytes, 0, head, 0, head.length);

            // 类名称，全名
            int length = Ints.fromBytes(bytes[0], bytes[1], bytes[2], bytes[3]);
            String classString = new String(bytes, 4, length);

            // 后面是实际的序列化数据
            byte[] data = new byte[bytes.length - length - 4];
            System.arraycopy(bytes, length + 4, data, 0, data.length);
            return deserialize(kryo, data, (Class<T>) Class.forName(classString));
        } catch (ClassNotFoundException e) {
            throw new IllegalStateException(e);
        } finally {
            // 将kryo对象归还到池中
            kryoPool.free(kryo);
        }
    }



    public T deserialize(Kryo kryo, byte[] bytes, Class<T> clazz) {
        ByteArrayInputStream in = new ByteArrayInputStream(bytes);
        Input input = new Input(in);
        return kryo.readObject(input, clazz);
    }
}