package backtype.storm.serialization;

import backtype.storm.utils.ListDelegate;
import com.esotericsoftware.kryo.ObjectBuffer;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.util.List;
import java.util.Map;

public class KryoValuesSerializer {
    ObjectBuffer _kryo;
    ListDelegate _delegate;
    
    public KryoValuesSerializer(Map conf) {
        _kryo = SerializationFactory.getKryo(conf);
        _delegate = new ListDelegate();
    }
    
    public void serializeInto(List<Object> values, OutputStream out) throws IOException {
        // this ensures that list of values is always written the same way, regardless
        // of whether it's a java collection or one of clojure's persistent collections 
        // (which have different serializers)
        // Doing this lets us deserialize as ArrayList and avoid writing the class here
        _delegate.setDelegate(values);
        _kryo.writeObject(out, _delegate);        
    }
    
    public byte[] serialize(List<Object> values) throws IOException {
        ByteArrayOutputStream out = new ByteArrayOutputStream();
        serializeInto(values, out);
        return out.toByteArray();
    }
    
    public byte[] serializeObject(Object obj) {
        ByteArrayOutputStream out = new ByteArrayOutputStream();
        _kryo.writeClassAndObject(out, obj);
        return out.toByteArray();
    }
}