/**
 *    Copyright 2014 Renren.com
 *
 *    Licensed under the Apache License, Version 2.0 (the "License");
 *    you may not use this file except in compliance with the License.
 *    You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 *    Unless required by applicable law or agreed to in writing, software
 *    distributed under the License is distributed on an "AS IS" BASIS,
 *    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *    See the License for the specific language governing permissions and
 *    limitations under the License.
 */
package com.renren.Wario.zookeeper;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.CountDownLatch;

import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.ZooKeeper.States;
import org.apache.zookeeper.data.Stat;

public class ZooKeeperClient implements Watcher {

	private static Logger logger = LogManager.getLogger(ZooKeeperClient.class
			.getName());

	private ZooKeeper zk = null;
	private final String connectionString;
	private final int sessionTimeout;
	private final String scheme = "digest";
	private final String auth;

	private volatile boolean isAvailable;
	private final String ZK_PATH;
	public ZooKeeperState state = null;

	private CountDownLatch countDownLatch = null;

	public ZooKeeperClient(String connectionString, int sessionTimeout) {
		this(connectionString, sessionTimeout, "");
	}

	public ZooKeeperClient(String connectionString, int sessionTimeout,
			String auth) {
		this.connectionString = connectionString;
		this.sessionTimeout = sessionTimeout;
		this.auth = auth;
		isAvailable = false;
		ZK_PATH = "/god_damn_zookeeper_" + connectionString;
		state = new ZooKeeperState(connectionString);
	}

	public void createConnection() {
		releaseConnection();
		countDownLatch = new CountDownLatch(1);

		try {
			zk = new ZooKeeper(connectionString, sessionTimeout, this);
			countDownLatch.await();
			if (!"".equals(auth)) {
				zk.addAuthInfo(scheme, auth.getBytes());
			}
		} catch (IOException e) {
			logger.error("Create connection failed! IOException occured!\n"
					+ e.toString());
		} catch (InterruptedException e) {
			logger.error("InterruptedException occured!\n" + e.toString());
		}

		try {
			zk.create(ZK_PATH, "this is a test node.".getBytes(),
					Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
		} catch (KeeperException e) {

		} catch (InterruptedException e) {

		}
	}

	public void releaseConnection() {
		isAvailable = false;
		if (zk != null) {
			try {
				zk.close();
			} catch (InterruptedException e) {
				logger.error("InterruptedException occured!\n" + e.toString());
			}
		}
	}

	@Override
	public void process(WatchedEvent event) {
		logger.info("SessionWatcher: " + connectionString + "|"
				+ event.getType() + "|" + event.getState());

		if (event.getType() == EventType.None) {
			if (event.getState().equals(KeeperState.SyncConnected)) {
				isAvailable = true;
				countDownLatch.countDown();
			} else if (event.getState().equals(KeeperState.Expired)) {
				isAvailable = false;
				createConnection();
			} else if (event.getState().equals(KeeperState.Disconnected)) {
				isAvailable = false;
				try {
					Thread.sleep(3000);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
				if (this.zk.getState() == States.CONNECTED) {
					isAvailable = true;
					return;
				}
				createConnection();
			}
		}
	}

	public boolean isAvailable() {
		return isAvailable;
	}

	/**
	 * @return the connectionString
	 */
	public String getConnectionString() {
		return connectionString;
	}

	/**
	 * @return the sessionTimeout
	 */
	public int getSessionTimeout() {
		return sessionTimeout;
	}

	public Stat exists(String path) throws KeeperException,
			InterruptedException {
		return zk.exists(path, false);
	}

	public byte[] getData(String path) throws KeeperException,
			InterruptedException {
		return zk.getData(path, false, null);
	}

	public List<String> getChildren(String path) throws KeeperException,
			InterruptedException {
		return zk.getChildren(path, false);
	}

	public Stat testExists(String path) throws KeeperException,
			InterruptedException {
		return zk.exists(ZK_PATH + path, false);
	}

	public void testCreate(String path, byte[] data) throws KeeperException,
			InterruptedException {
		zk.create(ZK_PATH + path, data, Ids.OPEN_ACL_UNSAFE,
				CreateMode.PERSISTENT);
	}

	public byte[] testGetData(String path) throws KeeperException,
			InterruptedException {
		return zk.getData(ZK_PATH + path, false, null);
	}

	public void testSetData(String path, byte[] data) throws KeeperException,
			InterruptedException {
		zk.setData(ZK_PATH + path, data, -1);
	}

	public List<String> testGetChildren(String path) throws KeeperException,
			InterruptedException {
		return zk.getChildren(ZK_PATH + path, false);
	}

	public void testDdelete(String path) throws InterruptedException,
			KeeperException {
		zk.delete(ZK_PATH + path, -1);
	}

}

@@ -2511,38 +2511,41 @@
 ion 
-failed! IOException occured!%5Cn
+%22 + connectionString + %22 failed! 
 %22%0A%09%09
@@ -2622,41 +2622,66 @@
 or(%22
-InterruptedException occured!%5Cn%22 
+Create connection %22 + connectionString + %22 failed! %22%0A%09%09%09%09%09
 + e.
@@ -2849,46 +2849,206 @@
 ) %7B%0A
-%0A%09%09%7D catch (InterruptedException e) %7B%0A
+%09%09%09logger.error(%22Create test node %22 + ZK_PATH + %22 faild! %22%0A%09%09%09%09%09+ e.toString());%0A%09%09%7D catch (InterruptedException e) %7B%0A%09%09%09logger.error(%22Create test node %22 + ZK_PATH + %22 faild! %22%0A%09%09%09%09%09+ e.toString());
 %0A%09%09%7D
@@ -3215,39 +3215,64 @@
 or(%22
-InterruptedException occured!%5Cn
+Close connection %22 + connectionString%0A%09%09%09%09%09%09+ %22 failed! 
 %22 + 

Format the log.