import org.apache.curator.framework.CuratorFramework;
import org.apache.log4j.Logger;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;

/**
 * 共享状态
 * Created on 16/11/12.
 */
public final class ClusterStateUtils {
	private Logger log = Logger.getLogger(ClusterStateUtils.class);

	private final String namespace = "/RCM";//命名空间
	private final String lockpath = "/lock";//锁
	private String businessType;//业务类型
	private Set<String> regions;
	private CuratorFramework client;


	private ClusterStateUtils() {

	}

	public ClusterStateUtils(String businessType, Set<String> regions,CuratorFramework client) {
		this.businessType = "/" + businessType;
		this.client = client;
		this.regions = regions;
		initWorkPath();
	}

	/**
	 * 从字节数组获取对象
	 *
	 * @EditTime 2007-8-13 上午11:46:34
	 */
	public static <T> Object getObjectFromBytes(byte[] objBytes,Class<T> cls) throws Exception {
		if (objBytes == null || objBytes.length == 0) {
			return null;
		}
		ByteArrayInputStream bi = new ByteArrayInputStream(objBytes);
		ObjectInputStream oi = new ObjectInputStream(bi);
		return (T) oi.readObject();
	}

	/**
	 * 从对象获取一个字节数组
	 *
	 * @EditTime 2007-8-13 上午11:46:56
	 */
	public static <T > byte[] getBytesFromObject(T obj) throws Exception {
		if (obj == null) {
			return null;
		}
		ByteArrayOutputStream bo = new ByteArrayOutputStream();
		ObjectOutputStream oo = new ObjectOutputStream(bo);
		oo.writeObject(obj);
		return bo.toByteArray();
	}

	/**
	 * 初始化工作目录
	 */
	private void initWorkPath() {
		try {
			safeCreatePath(namespace + businessType + lockpath);
			for (String region : regions) {
				safeCreatePath(namespace + businessType + "/" + region);
			}
		} catch (Exception e) {
			log.error("initWorkPath",e);
		}

	}

	private void safeCreatePath(String path) throws Exception {
		Stat stat = client.checkExists().forPath(path);
		if (stat == null) {
			client.create().creatingParentsIfNeeded().forPath(path);
		}
	}

	/**
	 * 获取锁
	 *
	 * @param lockName
	 * @return
	 */
	public boolean acquire(String lockName) {
		boolean success = false;
		try {
			client.create().withMode(CreateMode.EPHEMERAL).forPath(namespace + businessType + lockpath + "/" + lockName);
			success = true;
		} catch (Exception e) {
			log.error("acquire",e);
		}
		return success;
	}

	/**
	 * 释放锁
	 *
	 * @param lockName
	 * @return
	 */
	public boolean release(String lockName) {
		boolean success = false;
		try {
			client.delete().forPath(namespace + businessType + lockpath + "/" + lockName);
			success = true;
		} catch (Exception e) {
			log.error("release",e);
		}
		return success;
	}

	/**
	 * 重建目录
	 *
	 * @param region
	 * @return
	 */
	public boolean rebuildPath(String region) {
		if (!regions.contains(region) || region.equals(lockpath)) {
			return false;
		}
		boolean success = false;
		try {
			client.delete().deletingChildrenIfNeeded().forPath(namespace + businessType + "/" + region);
			client.create().forPath(namespace + businessType + "/" + region);
			success = true;
		} catch (Exception e) {
			log.error("rebuildPath",e);
		}
		return success;
	}

	/**
	 * 获取值
	 *
	 * @param region
	 * @param key
	 * @return
	 */
	public byte[] get(String region, String key) {
		byte[] bytes = null;
		try {
			bytes = client.getData().forPath(namespace + businessType + "/" + region + "/" + key);
		} catch (Exception e) {
			if (!region.contains("Try")) {
				log.error(region + key, e);
			}
		}
		return bytes;
	}

	/**
	 * 获取值
	 *
	 * @param region
	 * @param key
	 * @param targetClass
	 * @param <T>
	 * @return
	 */
	public <T> T get(String region, String key, Class<T> targetClass) {
		T obj = null;
		byte[] bytes = null;
		try {
			bytes = get(region, key);
			obj = (T) getObjectFromBytes(bytes, targetClass);
		} catch (Exception e) {
			if (!region.contains("Try")) {
				log.error(region + key, e);
			}
		}
		return obj;
	}

	/**
	 * 写入值
	 *
	 * @param region
	 * @param key
	 * @param bytes
	 * @return
	 */
	public boolean set(String region, String key, byte[] bytes) {
		boolean success = false;
		try {
			Stat stat = client.checkExists().forPath(namespace + businessType + "/" + region + "/" + key);
			if (stat != null) {
				client.delete().forPath(namespace + businessType + "/" + region + "/" + key);
			}
			client.create().forPath(namespace + businessType + "/" + region + "/" + key, bytes);
			success = true;
		} catch (Exception e) {
			log.error("set",e);
		}
		return success;
	}

	/**
	 * 写入值
	 *
	 * @param region
	 * @param key
	 * @param obj
	 * @param <T>
	 * @return
	 */
	public <T> boolean set(String region, String key, T obj) {
		boolean success = false;
		try {
			set(region, key, getBytesFromObject(obj));
			success = true;
		} catch (Exception e) {
			log.error("set",e);
		}
		return success;
	}

	/**
	 * 移除
	 *
	 * @param region
	 * @param key
	 * @return
	 */
	public boolean remove(String region, String key) {
		boolean success = false;
		try {
			client.delete().guaranteed().forPath(namespace + businessType + "/" + region + "/" + key);
			success = true;
		} catch (Exception e) {
			log.error("remove",e);
		}
		return success;
	}


	/**
	 * 获取子节点
	 *
	 * @param region
	 * @return
	 */
	public List<String> getChildren(String region) {
		List<String> keys = new ArrayList<String>();
		try {
			keys = client.getChildren().forPath(namespace + businessType + "/" + region);
		} catch (Exception e) {
			log.error("getChildren",e);
		}
		return keys;
	}
}