package org.craft.atom.protocol.rpc;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;

import org.craft.atom.protocol.ProtocolException;
import org.craft.atom.protocol.rpc.model.RpcBody;
import org.craft.atom.protocol.rpc.spi.Serialization;
import org.craft.atom.util.Assert;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;

/**
 * The implementor using <a href="https://github.com/EsotericSoftware/kryo">kryo</a>.
 * <p>
 * Not thread safe.
 * 
 * @author mindwind
 * @version 1.0, Jul 23, 2014
 */
public class KryoSerialization implements Serialization<RpcBody> {
	
	// singleton
	private static final KryoSerialization INSTNACE = new KryoSerialization();
	public static KryoSerialization getInstance() { return INSTNACE; } 
	private KryoSerialization() {}
	
	
	// thread local cache
    private static final ThreadLocal<Kryo> CACHE = new ThreadLocal<Kryo>() {
    	@Override
    	protected Kryo initialValue() {
            Kryo kryo = new Kryo();
            kryo.register(RpcBody.class);
            return kryo;
        }
    };
	
	@Override
	public byte type() {
		return 1;
	}

	@Override
	public byte[] serialize(RpcBody rb) {
		try {
			Assert.notNull(rb);
		    ByteArrayOutputStream baos = new ByteArrayOutputStream();
		    Output output = new Output(baos);
		    CACHE.get().writeObject(output, rb);
		    output.close();
		    return baos.toByteArray();
		} catch (Exception e) {
			throw new ProtocolException(e);
		}
	}

	@Override
	public RpcBody deserialize(byte[] bytes) {
		return deserialize(bytes, 0);
	}

	@Override
	public RpcBody deserialize(byte[] bytes, int off) {
		try {
		    Assert.notNull(bytes);
		    ByteArrayInputStream bais = new ByteArrayInputStream(bytes, off, bytes.length - off);
		    Input input = new Input(bais);
		    RpcBody rb = CACHE.get().readObject(input, RpcBody.class);
	        input.close();
		    return rb;
		} catch (Exception e) {
			throw new ProtocolException(e);
		}
	}

}

@@ -105,16 +105,52 @@
 tStream;
+%0Aimport java.lang.ref.SoftReference;
 %0A%0Aimport
@@ -959,21 +959,36 @@
 adLocal%3C
+SoftReference%3C
 Kryo%3E
+%3E
  CACHE =
@@ -1004,21 +1004,36 @@
 adLocal%3C
+SoftReference%3C
 Kryo%3E
+%3E
 () %7B%0A   
@@ -1059,20 +1059,35 @@
 otected 
+SoftReference%3C
 Kryo
+%3E
  initial
@@ -1193,20 +1193,45 @@
  return 
+new SoftReference%3CKryo%3E(
 kryo
+)
 ;%0A      
@@ -1483,25 +1483,20 @@
 ;%0A%09%09    
-CACHE.get
+kryo
 ().write
@@ -1637,16 +1637,185 @@
 ;%0A%09%09%7D%0A%09%7D
+%0A%09%0A%09private Kryo kryo() %7B%0A%09%09Kryo kryo = CACHE.get().get();%0A%09%09if (kryo == null) %7B%0A%09%09%09kryo = new Kryo();%0A%09%09%09CACHE.set(new SoftReference%3CKryo%3E(kryo));%0A%09%09%7D%0A%09%09return kryo;%0A%09%7D
 %0A%0A%09@Over
@@ -2148,25 +2148,20 @@
 dy rb = 
-CACHE.get
+kryo
 ().readO

improve - use SoftReference