import java.nio.ByteBuffer;

import org.springframework.data.redis.serializer.RedisElementReader;
import org.springframework.data.redis.serializer.RedisElementWriter;
import org.springframework.data.redis.serializer.RedisSerializationContext.SerializationPair;
import org.springframework.stereotype.Component;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.ByteBufferInputStream;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import com.esotericsoftware.kryo.pool.KryoPool;

@Component
public class KryoObjectSerializer implements SerializationPair<Object> {
    /** Kryo池 */
    KryoPool kryoPool = new KryoPool.Builder(Kryo::new).build();

    private Kryo getKryo() {
        Kryo kryo = kryoPool.borrow();
        // 設定類載入器，應對ClassCastException異常
        // 如果不設定的話，使用spring-boot-devtools時，會拋異常
        kryo.setClassLoader(getClass().getClassLoader());
        return kryo;
    }

    @Override
    public RedisElementReader<Object> getReader() {
        return new RedisElementReader<Object>() {
            @Override
            public Object read(ByteBuffer buffer) {
                Kryo kryo = getKryo();
                try {
                    return kryo.readClassAndObject(
                            new Input(new ByteBufferInputStream(buffer)));
                } finally {
                    kryoPool.release(kryo);
                }
            }
        };
    }

    @Override
    public RedisElementWriter<Object> getWriter() {
        return new RedisElementWriter<Object>() {
            @Override
            public ByteBuffer write(Object element) {
                Kryo kryo = getKryo();
                Output output = new Output(1024, 1024 * 100);
                try {
                    kryo.writeClassAndObject(output, element);
                } finally {
                    kryoPool.release(kryo);
                }
                return ByteBuffer.wrap(output.getBuffer());
            }
        };
    }
}