package com.XXX.currencyrate.atomic.connector;

import java.io.ByteArrayOutputStream;
import java.nio.ByteBuffer;
import java.nio.charset.Charset;

import org.joda.time.DateTime;
import org.joda.time.LocalDate;
import org.joda.time.LocalTime;
import org.joda.time.PeriodType;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import com.esotericsoftware.kryo.pool.KryoCallback;
import com.esotericsoftware.kryo.pool.KryoFactory;
import com.esotericsoftware.kryo.pool.KryoPool;
import com.esotericsoftware.kryo.serializers.JavaSerializer;
import com.lambdaworks.redis.codec.RedisCodec;
import com.XXX.schema.utility.CurrencyRateConversion;

public class KryoCodec implements RedisCodec<String, CurrencyRateConversion> {
	private Charset charset = Charset.forName("UTF-8");

	private KryoFactory factory = new KryoFactory() {
		public Kryo create() {
			Kryo kryo = new Kryo();

			// Kryo can't marshal joda, so we use plain java serializer.
			kryo.register(LocalDate.class, new JavaSerializer());
			kryo.register(LocalTime.class, new JavaSerializer());
			kryo.register(DateTime.class, new JavaSerializer());
			kryo.register(PeriodType.class, new JavaSerializer());
			return kryo;
		}
	};

	private KryoPool pool = new KryoPool.Builder(factory).softReferences().build();

	@Override
	public String decodeKey(ByteBuffer bytes) {
		return charset.decode(bytes).toString();
	}

	@Override
	public CurrencyRateConversion decodeValue(final ByteBuffer byteBuffer) {
		final byte[] array = new byte[byteBuffer.remaining()];
		byteBuffer.get(array);
		CurrencyRateConversion currencyRateConversion = pool.run(new KryoCallback<CurrencyRateConversion>() {
			public CurrencyRateConversion execute(Kryo kryo) {
				return kryo.readObject(new Input(array), CurrencyRateConversion.class);
			}
		});
		return currencyRateConversion;
	}

	@Override
	public ByteBuffer encodeKey(String key) {
		return charset.encode(key);
	}

	@Override
	public ByteBuffer encodeValue(CurrencyRateConversion value) {
		Output output = new Output(new ByteArrayOutputStream());
		Kryo kryo = pool.borrow();
		kryo.writeObject(output, value);
		output.close();
		pool.release(kryo);
		return ByteBuffer.wrap(((ByteArrayOutputStream) output.getOutputStream()).toByteArray());
	}

}