package com.sdyc.ndmp.protobuf.serializer;


import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import com.google.common.collect.ImmutableMap;
import com.google.protobuf.GeneratedMessage;
import com.sdyc.ndmp.protobuf.dubbo.support.KryoFactory;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.util.Map;

/**
 * <pre>
 *
 * Created by IntelliJ IDEA.
 * User: zhenqin
 * Date: 14/12/10
 * Time: 09:27
 * To change this template use File | Settings | File Templates.
 *
 * </pre>
 *
 * @author zhenqin
 */
public class KryoSerializer<T> implements Serializer<T> {


    protected final Kryo kryo;


    protected final Class<T> objectClass;


    public KryoSerializer(Class<T> objectClass) {
        this(objectClass,
                ImmutableMap.<Class<?>, com.esotericsoftware.kryo.Serializer>of(
                GeneratedMessage.class, new ProtobufKryoSerializer()));
    }



    public KryoSerializer(Class<T> objectClass, Map<Class<?>, com.esotericsoftware.kryo.Serializer> classMap) {
        kryo = new Kryo();
        for (Map.Entry<Class<?>, com.esotericsoftware.kryo.Serializer> entry : classMap.entrySet()) {
            kryo.register(entry.getKey(), entry.getValue());
        }
        this.objectClass = objectClass;
    }



    protected KryoSerializer(Kryo kryo, Class<T> objectClass) {
        this.kryo = kryo;
        this.objectClass = objectClass;
    }



    public static <S> KryoSerializer<S> getKryoSerializer(Class<S> objectClass) {
        return new KryoSerializer<S>(KryoFactory.newInstance(), objectClass);
    }




    public static <S> KryoSerializer<S> getKryoSerializer(KryoFactory.Call call, Class<S> objectClass) {
        return new KryoSerializer<S>(KryoFactory.newInstance(call), objectClass);
    }



    @Override
    public byte[] serialize(T o) {
        ByteArrayOutputStream out = new ByteArrayOutputStream();
        Output output = new Output(out);
        kryo.writeObject(output, o);
        output.flush();
        return out.toByteArray();
    }



    @Override
    public T deserialize(byte[] bytes) {
        return deserialize(bytes, this.objectClass);
    }



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

}