package cascading.kryo;
import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import org.apache.hadoop.io.serializer.Deserializer;
import java.io.DataInputStream;
import java.io.IOException;
import java.io.InputStream;
public class KryoDeserializer implements Deserializer<Object> {
private Kryo kryo;
private final KryoSerialization kryoSerialization;
private final Class<Object> klass;
private DataInputStream inputStream;
public KryoDeserializer(KryoSerialization kryoSerialization, Class<Object> klass) {
this.kryoSerialization = kryoSerialization;
this.klass = klass;
}
public void open(InputStream in) throws IOException {
kryo = kryoSerialization.populatedKryo();
if( in instanceof DataInputStream)
inputStream = (DataInputStream) in;
else
inputStream = new DataInputStream( in );
}
public Object deserialize(Object o) throws IOException {
byte[] bytes = new byte[inputStream.readInt()];
inputStream.readFully( bytes );
return kryo.readObject(new Input(bytes), klass);
}
// TODO: Bump the kryo version, add a kryo.reset();
public void close() throws IOException {
try {
if( inputStream != null )
inputStream.close();
} finally {
inputStream = null;
}
}
}