package com.imadcn.framework.idworker.register.zookeeper;

import java.util.Date;
import java.util.List;
import java.util.concurrent.TimeUnit;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.framework.state.ConnectionStateListener;
import org.jboss.netty.handler.timeout.TimeoutException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.imadcn.framework.idworker.config.ApplicationConfiguration;
import com.imadcn.framework.idworker.exception.RegException;
import com.imadcn.framework.idworker.register.WorkerRegister;
import com.imadcn.framework.idworker.registry.CoordinatorRegistryCenter;

/**
 * 机器信息注册
 * 
 * @author yangchao
 * @since 2017-10-18
 */
public class ZookeeperWorkerRegister implements WorkerRegister {

	private static final Logger logger = LoggerFactory.getLogger(ZookeeperWorkerRegister.class);

	/**
	 * 最大机器数
	 */
	private static final long MAX_WORKER_NUM = 1024;
	/**
	 * 加锁最大等待时间
	 */
	private static final int MAX_LOCK_WAIT_TIME_MS = 30 * 1000;

	private final CoordinatorRegistryCenter regCenter;
	/**
	 * zk节点信息
	 */
	private final NodePath nodePath;

	public ZookeeperWorkerRegister(CoordinatorRegistryCenter regCenter, ApplicationConfiguration applicationConfiguration) {
		this.regCenter = regCenter;
		this.nodePath = new NodePath(applicationConfiguration.getGroup());
	}

	/**
	 * 向zookeeper注册workerId
	 * 
	 * @return workerId workerId
	 */
	@Override
	public long register() {
		CuratorFramework client = (CuratorFramework) regCenter.getRawClient();
		InterProcessMutex lock = new InterProcessMutex(client, nodePath.getGroupPath());
		try {
			int numOfChildren = regCenter.getNumChildren(nodePath.getWorkerPath());
			if (numOfChildren < MAX_WORKER_NUM) {
				if (!lock.acquire(MAX_LOCK_WAIT_TIME_MS, TimeUnit.MILLISECONDS)) {
					String message = String.format("acquire lock failed after %s ms.", MAX_LOCK_WAIT_TIME_MS);
					throw new TimeoutException(message);
				}
				long sessionId = client.getZookeeperClient().getZooKeeper().getSessionId();
				logger.debug("current session id is : {}", sessionId);
				List<String> children = regCenter.getChildrenKeys(nodePath.getWorkerPath());
				for (int workerId = 0; workerId < MAX_WORKER_NUM; workerId++) {
					String workderIdStr = String.valueOf(workerId);
					String key = nodePath.getWorkerPath() + "/" + workerId;
					NodeInfo currentNode = new NodeInfo(sessionId, workerId);
					currentNode.setCreateTime(new Date());
					currentNode.setUpdateTime(new Date());
					if (!children.contains(workderIdStr)) {
						String value = jsonizeNodeInfo(currentNode);
						logger.info("snowflake worker register with : {}", value);
						regCenter.persistEphemeral(key, value);
						// 将workerId保存进nodePath 
						nodePath.setWorkerId(workerId);
						return workerId;
					} else {
						String value = regCenter.get(key);
						NodeInfo cacheNode = JSON.parseObject(value, NodeInfo.class);
						if (currentNode.equals(cacheNode)) {
							logger.info("use cached nodeinfo, {}", value);
							cacheNode.setUpdateTime(new Date());
							regCenter.persistEphemeral(key, jsonizeNodeInfo(cacheNode));
							return cacheNode.getWorkerId();
						}
					}
				}
			}
			throw new RegException("max worker num reached. register failed");
		} catch (Exception e) {
			logger.error("", e);
			throw new IllegalStateException(e.getMessage(), e);
		} finally {
			try {
				lock.release();
			} catch (Exception ignored) {
				logger.error("", ignored);
			}
		}
	}
	
	/**
	 * 添加连接监听
	 * @param listener zk状态监听listener
	 */
	public void addConnectionLJistener(ConnectionStateListener listener) {
		CuratorFramework client = (CuratorFramework) regCenter.getRawClient();
		client.getConnectionStateListenable().addListener(listener);
	}
	
	/**
	 * 关闭注册
	 */
	@Override
	public synchronized void logout() {
		// 移除注册节点
		regCenter.remove(nodePath.getWorkerIdPath());
		// 关闭连接
		regCenter.close();
	}
	
	private String jsonizeNodeInfo(NodeInfo nodeInfo) {
		String dateFormat = "yyyy-MM-dd HH:mm:ss";
		return JSON.toJSONStringWithDateFormat(nodeInfo, dateFormat, SerializerFeature.WriteDateUseDateFormat);
	}
}

@@ -191,16 +191,80 @@
 mework;%0A
+import org.apache.curator.framework.imps.CuratorFrameworkState;%0A
 import o
@@ -4060,16 +4060,150 @@
 out() %7B%0A
+%09%09CuratorFramework client = (CuratorFramework) regCenter.getRawClient();%0A%09%09if (client.getState() == CuratorFrameworkState.STARTED) %7B%0A%09
 %09%09// %E7%A7%BB%E9%99%A4%E6%B3%A8
@@ -4206,16 +4206,17 @@
  %E7%A7%BB%E9%99%A4%E6%B3%A8%E5%86%8C%E8%8A%82%E7%82%B9%0A
+%09
 %09%09regCen
@@ -4261,16 +4261,17 @@
 ;%0A%09%09
+%09
 // %E5%85%B3%E9%97%AD%E8%BF%9E%E6%8E%A5%0A
 %09%09re
@@ -4266,16 +4266,17 @@
 // %E5%85%B3%E9%97%AD%E8%BF%9E%E6%8E%A5%0A
+%09
 %09%09regCen
@@ -4288,16 +4288,20 @@
 lose();%0A
+%09%09%7D%0A
 %09%7D%0A%09%0A%09pr

fix bug: instance must be started before calling this method