package omq.common.util;

import java.io.IOException;
import java.util.Properties;

import org.apache.log4j.Logger;

import omq.common.broker.Broker;
import omq.common.event.Event;
import omq.common.message.Request;
import omq.common.message.Response;
import omq.common.util.Serializers.GsonImp;
import omq.common.util.Serializers.ISerializer;
import omq.common.util.Serializers.JavaImp;
import omq.common.util.Serializers.KryoImp;
import omq.exception.SerializerException;
import omq.server.RemoteObject;

/**
 * 
 * @author Sergi Toda <sergi.toda@estudiants.urv.cat>
 * 
 */
public class Serializer {
	private static final Logger logger = Logger.getLogger(Serializer.class.getName());
	public static String kryo = "kryo";
	public static String java = "java";
	public static String gson = "gson";

	// Client serializer
	public static ISerializer serializer;

	// Server serializers
	private static ISerializer kryoSerializer;
	private static ISerializer javaSerializer;
	private static ISerializer gsonSerializer;

	private static Boolean getEnableCompression() {
		Properties env = Broker.getEnvironment();
		return Boolean.valueOf(env.getProperty(ParameterQueue.ENABLECOMPRESSION, "false"));
	}

	public static ISerializer getInstance() throws SerializerException {
		if (serializer == null) {
			try {
				Properties env = Broker.getEnvironment();
				String className = env.getProperty(ParameterQueue.SERIALIZER_NAME, Serializer.java);

				if (className == null || className.isEmpty()) {
					throw new ClassNotFoundException("Class name is null or empty.");
				}

				serializer = getInstance(className);
			} catch (Exception ex) {
				throw new SerializerException(ex.getMessage(), ex);
			}
		}

		return serializer;
	}

	public static ISerializer getInstance(String type) throws SerializerException {
		if (kryo.equals(type)) {
			if (kryoSerializer == null) {
				kryoSerializer = new KryoImp();
			}
			return kryoSerializer;
		} else if (gson.endsWith(type)) {
			if (gsonSerializer == null) {
				gsonSerializer = new GsonImp();
			}
			return gsonSerializer;
		} else {
			if (javaSerializer == null) {
				javaSerializer = new JavaImp();
			}
			return javaSerializer;
		}
	}

	public static byte[] serialize(String type, Object obj) throws SerializerException {
		ISerializer instance = getInstance(type);

		Boolean enableCompression = getEnableCompression();
		if (enableCompression) {
			byte[] objSerialized = instance.serialize(obj);
			try {
				return Zipper.zip(objSerialized);
			} catch (IOException e) {
				throw new SerializerException(e.getMessage(), e);
			}
		} else {
			return instance.serialize(obj);
		}
	}

	// TODO: remove this function and think about the event serialization
	public static byte[] serialize(Object obj) throws SerializerException {
		ISerializer instance = getInstance();

		Boolean enableCompression = getEnableCompression();
		if (enableCompression) {
			byte[] objSerialized = instance.serialize(obj);
			try {
				return Zipper.zip(objSerialized);
			} catch (IOException e) {
				throw new SerializerException(e.getMessage(), e);
			}
		} else {
			return instance.serialize(obj);
		}
	}

	public static Request deserializeRequest(String type, byte[] bytes, RemoteObject obj) throws SerializerException {
		ISerializer instance = getInstance(type);

		Boolean enableCompression = getEnableCompression();
		if (enableCompression) {
			try {
				byte[] unZippedBytes = Zipper.unzip(bytes);
				return instance.deserializeRequest(unZippedBytes, obj);
			} catch (IOException e) {
				throw new SerializerException(e.getMessage(), e);
			}
		} else {
			return instance.deserializeRequest(bytes, obj);
		}
	}

	public static Response deserializeResponse(byte[] bytes, Class<?> type) throws SerializerException {
		ISerializer instance = getInstance();

		Boolean enableCompression = getEnableCompression();
		if (enableCompression) {
			try {
				byte[] unZippedBytes = Zipper.unzip(bytes);
				return instance.deserializeResponse(unZippedBytes, type);
			} catch (IOException e) {
				throw new SerializerException(e.getMessage(), e);
			}
		} else {
			return instance.deserializeResponse(bytes, type);
		}
	}

	public static Event deserializeEvent(byte[] bytes) throws SerializerException {
		ISerializer instance = getInstance();

		Boolean enableCompression = getEnableCompression();
		if (enableCompression) {
			try {
				byte[] unZippedBytes = Zipper.unzip(bytes);
				return instance.deserializeEvent(unZippedBytes);
			} catch (IOException e) {
				throw new SerializerException(e.getMessage(), e);
			}
		} else {
			return instance.deserializeEvent(bytes);
		}
	}

	public static void removeSerializers() {
		logger.warn("Removing serializers");
		serializer = null;
		kryoSerializer = null;
		javaSerializer = null;
		gsonSerializer = null;
	}
}