content: /**
   * Copyright 2010 the original author or authors.
   *
   * 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.github.zkclient;
  
  import java.io.IOException;
  import java.lang.reflect.InvocationTargetException;
  import java.lang.reflect.Method;
  import java.util.List;
  import java.util.concurrent.locks.Lock;
  import java.util.concurrent.locks.ReentrantLock;
  
  import org.apache.zookeeper.CreateMode;
  import org.apache.zookeeper.KeeperException;
  import org.apache.zookeeper.Watcher;
  import org.apache.zookeeper.ZooKeeper;
  import org.apache.zookeeper.ZooDefs.Ids;
  import org.apache.zookeeper.ZooKeeper.States;
  import org.apache.zookeeper.data.Stat;
  
  import com.github.zkclient.exception.ZkException;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  public class ZkConnection {
  
      private static final Logger LOG = LoggerFactory.getLogger(ZkConnection.class);
  
      private ZooKeeper _zk = null;
      private final Lock _zookeeperLock = new ReentrantLock();
  
      private final String _servers;
      private final int _sessionTimeOut;
  
      private static final Method method;
  
      static {
          Method[] methods = ZooKeeper.class.getDeclaredMethods();
          Method m = null;
          for (Method method : methods) {
              if (method.getName().equals("multi")) {
                  m = method;
                  break;
              }
          }
          method = m;
      }
  
      /**
       * build a zookeeper connection
       * @param zkServers      zookeeper connection string
       * @param sessionTimeOut session timeout in milliseconds
       */
      public ZkConnection(String zkServers, int sessionTimeOut) {
          _servers = zkServers;
          _sessionTimeOut = sessionTimeOut;
      }
  
      public void connect(Watcher watcher) {
          _zookeeperLock.lock();
          try {
              if (_zk != null) {
                  throw new IllegalStateException("zk client has already been started");
              }
              try {
                  LOG.debug("Creating new ZookKeeper instance to connect to " + _servers + ".");
                  _zk = new ZooKeeper(_servers, _sessionTimeOut, watcher);
              } catch (IOException e) {
                  throw new ZkException("Unable to connect to " + _servers, e);
              }
          } finally {
              _zookeeperLock.unlock();
          }
      }
  
      public void close() throws InterruptedException {
          _zookeeperLock.lock();
          try {
              if (_zk != null) {
                  LOG.debug("Closing ZooKeeper connected to " + _servers);
                  _zk.close();
                  _zk = null;
              }
          } finally {
              _zookeeperLock.unlock();
          }
      }
  
      public String create(String path, byte[] data, CreateMode mode) throws KeeperException, InterruptedException {
          return _zk.create(path, data, Ids.OPEN_ACL_UNSAFE, mode);
      }
  
      public void delete(String path) throws InterruptedException, KeeperException {
          _zk.delete(path, -1);
      }
  
      public boolean exists(String path, boolean watch) throws KeeperException, InterruptedException {
          return _zk.exists(path, watch) != null;
      }
  
      public List<String> getChildren(final String path, final boolean watch) throws KeeperException, InterruptedException {
          return _zk.getChildren(path, watch);
      }
  
      public byte[] readData(String path, Stat stat, boolean watch) throws KeeperException, InterruptedException {
          return _zk.getData(path, watch, stat);
      }
  
      /**
       * wrapper for 3.3.x/3.4.x
       *
       * @param ops multi operations
       * @return OpResult list
       */
      @SuppressWarnings("unchecked")
      public List<?> multi(Iterable<?> ops) {
          if (method == null) throw new UnsupportedOperationException("multi operation must use zookeeper 3.4+");
          try {
              return (List<?>) method.invoke(_zk, ops);
          } catch (IllegalArgumentException e) {
              throw new UnsupportedOperationException("ops must be 'org.apache.zookeeper.Op'");
          } catch (IllegalAccessException e) {
              throw new UnsupportedOperationException(e);
          } catch (InvocationTargetException e) {
              throw new RuntimeException(e);
          }
      }
  
      public Stat writeData(String path, byte[] data, int version) throws KeeperException, InterruptedException {
          return _zk.setData(path, data, version);
      }
  
      public States getZookeeperState() {
          return _zk != null ? _zk.getState() : null;
      }
  
      public long getCreateTime(String path) throws KeeperException, InterruptedException {
          Stat stat = _zk.exists(path, false);
          if (stat != null) {
              return stat.getCtime();
          }
          return -1;
      }
  
      public String getServers() {
          return _servers;
      }
  
      public ZooKeeper getZooKeeper() {
          return _zk;
      }
  }
  