def setUp( self ):
     zktestbase.TestBase.setUp(self)
     try:
         zookeeper.create(self.handle, "/zk-python-existstest","existstest", [ZOO_OPEN_ACL_UNSAFE],zookeeper.EPHEMERAL)
         zookeeper.create(self.handle, "/zk-python-aexiststest","existstest",[ZOO_OPEN_ACL_UNSAFE],zookeeper.EPHEMERAL)
     except:
         pass

    def register_consumer(self, consumer_group, consumer_id, topic):
        """Creates the following permanent node, if it does not exist already:
            /consumers/{consumer_group}/ids

        The data written at this node is just the consumer_id so that we can 
        later track who created what.

        We then create the following emphemeral node:
            /consumers/{consumer_group}/ids/{consumer_id}
        
        The data written in this node is a dictionary of topic names (in 
        unicode) to the number of threads that this consumer has registered
        for this topic (in our case, always one).
        """
        self._create_path_if_needed(self.path_for_consumer_ids(consumer_group),
                                    consumer_id)
        # Create an emphermal node for this consumer
        consumer_id_path = self.path_for_consumer_id(consumer_group, consumer_id)
        log.info("Registering Consumer {0}, trying to create {1}"
                 .format(consumer_id, consumer_id_path))
        zookeeper.create(self._zk.handle, 
                         consumer_id_path,
                         json.dumps({topic : 1}), # topic : # of threads
                         ZKUtil.ACL,
                         zookeeper.EPHEMERAL)

  def __tryGC(self, app_id, app_path):
    try:
#      print "try to obtrain app %s lock" % app_id
      val = zookeeper.get(self.handle, PATH_SEPARATOR.join([app_path, GC_TIME_PATH]), None)[0]
      lasttime = float(val)
    except zookeeper.NoNodeException:
      lasttime = 0
    if lasttime + GC_INTERVAL < time.time():
      # try to gc this app.
      gc_path = PATH_SEPARATOR.join([app_path, GC_LOCK_PATH])
      try:
        now = str(time.time())
        zookeeper.create(self.handle, gc_path, now, ZOO_ACL_OPEN, zookeeper.EPHEMERAL)
        # succeed to obtain lock.
        # TODO: should we handle the timeout of gc also?
        try:
          self.__executeGC(app_id, app_path)
          # update lasttime when gc was succeeded.
          now = str(time.time())
          self.__updateNode(PATH_SEPARATOR.join([app_path, GC_TIME_PATH]), now)
        except Exception, e:
          print "warning: gc error %s" % e
          traceback.print_exc()
        zookeeper.delete(self.handle, gc_path, -1)
      except zookeeper.NodeExistsException:
        # fail to obtain lock. try next time.
        pass

    def __init__(self, barriername, number_of_workers):
	self.cv = threading.Condition()
	self.connected = False
	self.barrier = "/" + barriername
	self.workers = number_of_workers
	zookeeper.set_log_stream(open('/dev/null'))
	def watcher(handle, type, state, path):
	    self.cv.acquire()
	    self.connected = True
	    self.cv.notify()
	    self.cv.release()

	self.cv.acquire()
	self.handle = zookeeper.init("localhost:2181", watcher, 10000)
	self.cv.wait(10.0)
	if not self.connected:
            print "Connection to ZooKeeper cluster timed out - is a server running on localhost:2181?"
            sys.exit()
	self.cv.release()
	try:
	    zookeeper.create(self.handle, self.barrier, '\x00', [ZOO_OPEN_ACL_UNSAFE], 0)
	except zookeeper.NodeExistsException:
	    pass
	except Exception, ex:
	    print ex
	    raise ex

def connect(host=None):
    global conn_cv, connected, handle
    global reconnect_host

    if host is None:
        host = "localhost:2181"

    reconnect_host = host

    conn_cv.acquire()
    handle = zookeeper.init(host, my_connection_watcher, 10000)
    while not connected:
        conn_cv.wait()
    conn_cv.release()

    connected = False
    while not connected:
        try:
            zookeeper.create(handle, ROOT, "1", [ZOO_OPEN_ACL_UNSAFE], 0)
            connected = True
        except zookeeper.NodeExistsException as e:
            # No worries
            connected = True
        except zookeeper.ConnectionLossException:
            continue
        except:
            raise

    def _create_path_if_needed(self, path, data=None):
        """Creates permanent nodes for all elements in the path if they don't
        already exist. Places data for each node created. (You'll probably want
        to use the consumer_id for this, just for accounting purposes -- it's 
        not used as part of the balancing algorithm).

        Our zc.zk.ZooKeeper object doesn't know how to create nodes, so we
        have to dig into the zookeeper base library. Note that we can't create
        all of it in one go -- ZooKeeper only allows atomic operations, and
        so we're creating these one at a time.
        """
        if not path.startswith("/"):
            raise ValueError("Paths must be fully qualified (begin with '/').")

        def _build_path(nodes):
            return "/" + "/".join(nodes)

        nodes_to_create = path[1:].split("/") # remove beginning "/"
        created_so_far = []
        for node in nodes_to_create:
            created_path = _build_path(created_so_far)
            if node and node not in self._zk.children(created_path).data:
                node_to_create = _build_path(created_so_far + [node])
                # If data is a string, we'll initialize the node with it...
                if isinstance(data, basestring):
                    init_data = data 
                else:
                    init_data = json.dumps(data)
                zookeeper.create(self._zk.handle, node_to_create, init_data, ZKUtil.ACL)
            created_so_far.append(node)

 def __init__(self,queuename):
     self.connected = False
     self.queuename = "/" + queuename
     self.cv = threading.Condition()
     zookeeper.set_log_stream(open("/dev/null"))        
     def watcher(handle,type,state,path):
         print "Connected"
         self.cv.acquire()
         self.connected = True
         self.cv.notify()
         self.cv.release()
         
     self.cv.acquire()            
     self.handle = zookeeper.init("localhost:2181", watcher, 10000, 0)
     self.cv.wait(10.0)
     if not self.connected:
         print "Connection to ZooKeeper cluster timed out - is a server running on localhost:2181?"
         sys.exit()
     self.cv.release()
     try:
         zookeeper.create(self.handle,self.queuename,"queue top level", [ZOO_OPEN_ACL_UNSAFE],0)
     except IOError, e:
         if e.message == zookeeper.zerror(zookeeper.NODEEXISTS):
             print "Queue already exists"    
         else:
             raise e

  def __init__(self,queuename, port, is_producer=False):
    self.connected = False
    self.queuename = "/" + queuename
    self.cv = threading.Condition()
    zookeeper.set_log_stream(open("/dev/null"))
    def watcher(handle,type,state,path):
      print "Connected"
      self.cv.acquire()
      self.connected = True
      self.cv.notify()
      self.cv.release()

    self.cv.acquire()
    self.handle = zookeeper.init("localhost:%d" % port, watcher, 10000)
    self.cv.wait(10.0)
    if not self.connected:
      print "Connection to ZooKeeper cluster timed out - is a server running on localhost:%d?" % port
      sys.exit()
    self.cv.release()
    if is_producer:
      while True:
        try:
          zookeeper.create(self.handle,self.queuename,"queue top level", [ZOO_OPEN_ACL_UNSAFE],0)
          print "Fila criada, OK"
          return
        except zookeeper.NodeExistsException:
          print "Tratorando filas existentes"
          while True:
            children = sorted(zookeeper.get_children(self.handle, self.queuename,None))
            if len(children) == 0:
              (data,stat) = zookeeper.get(self.handle, self.queuename, None)
              zookeeper.delete(self.handle, self.queuename, stat["version"])
              break
            for child in children:
              data = self.get_and_delete(self.queuename + "/" + child)

    def __init__(self, zkhosts, root=NODE_HQ_ROOT, alivenode='alive'):
        """zkhosts: a string or a list. list will be ','.join-ed into a string.
        root: root node path (any parents must exist, if any)
        """
        
        if not isinstance(zkhosts, basestring):
            zkhosts = ','.join(zkhosts)
        self.zkhosts = zkhosts
        self.ROOT = root
        self.alivenode = alivenode
        self.nodename = os.uname()[1]

        self.NODE_SERVERS = self.ROOT + '/servers'
        self.NODE_ME = self.NODE_SERVERS+'/'+self.nodename

        self.zh = zk.init(self.zkhosts, self.__watcher)

        self.jobs = {}

        self.initialize_node_structure()

        if not zk.exists(self.zh, self.NODE_ME):
            zk.create(self.zh, self.NODE_ME, '', PERM_WORLD)

        # setup notification
        zk.get_children(self.zh, self.NODE_SERVERS, self.__servers_watcher)
        #zk.get_children(self.zh, self.NODE_ME, self.__watcher)

        self.NODE_JOBS = self.NODE_ME+'/jobs'
        zk.acreate(self.zh, self.NODE_JOBS, '', PERM_WORLD)

def main():
	from optparse import OptionParser as OptionParser

	parser = OptionParser()
	parser.add_option('-v', '--verbose', action="store_true", dest="flag_verbose",
		help='Triggers debugging traces')

	(options, args) = parser.parse_args(sys.argv)

	# Logging configuration
	if options.flag_verbose:
		logging.basicConfig(
			format='%(asctime)s %(message)s',
			datefmt='%m/%d/%Y %I:%M:%S',
			level=logging.DEBUG)
	else:
		logging.basicConfig(
			format='%(asctime)s %(message)s',
			datefmt='%m/%d/%Y %I:%M:%S',
			level=logging.INFO)

	if len(args) < 2:
		raise ValueError("not enough CLI arguments")

	ns = args[1]
	cnxstr = load_config().get(ns, 'zookeeper')

	zookeeper.set_debug_level(zookeeper.LOG_LEVEL_INFO)
	zh = zookeeper.init(cnxstr)
	zookeeper.create(zh, PREFIX, '', acl_openbar, 0)
	create_tree(zh, boot_tree())
	init_namespace(zh, ns)
	zookeeper.close(zh)

  def __init__(self, service_name, connect_string, timeout=DEFAULT_TIMEOUT, default_port=6664):
    self.SERVICE_NODE = "/" + service_name
    self.AVAILABILITY_NODE = self.SERVICE_NODE + "/available"
    self.MEMBERSHIP_NODE = self.SERVICE_NODE + "/members"
    self.connected = False
    self.timeout = timeout
    self.default_port = default_port
    self.conn_cv = threading.Condition()
    self.conn_cv.acquire()
    self.handle = zookeeper.init(connect_string, self.connection_watcher, timeout)
    self.conn_cv.wait(timeout / 1000)
    self.conn_cv.release()
    self.watcher_lock = threading.Lock()
    self.logger = logging.getLogger("sincc")

    if not self.connected:
      raise SinClusterClientError("Unable to connect to %s" % connect_string)

    for path in [self.SERVICE_NODE, self.AVAILABILITY_NODE, self.MEMBERSHIP_NODE]:
      if not zookeeper.exists(self.handle, path):
        zookeeper.create(self.handle, path, "", [ZOO_OPEN_ACL_UNSAFE], 0)
    self.listeners = []
    # Start to watch both /members and /available
    zookeeper.get_children(self.handle, self.MEMBERSHIP_NODE, self.watcher)
    available = zookeeper.get_children(self.handle, self.AVAILABILITY_NODE, self.watcher)
    self.available_nodes = {}
    for node_id in available:
      self.available_nodes[int(node_id)] = Node(int(node_id),
                                                zookeeper.get(self.handle, self.AVAILABILITY_NODE + "/" + node_id)[0])

 def watch_session(handle, event_type, state, path):
     assert event_type == zookeeper.SESSION_EVENT
     assert not path
     if state == zookeeper.CONNECTED_STATE:
         if self.handle is None:
             self.handle = handle
             for watch in self.watches.clear():
                 self._watch(watch)
             for path, data in list(self.ephemeral.items()):
                 zookeeper.create(self.handle, path, data['data'],
                                  data['acl'], data['flags'])
         else:
             assert handle == self.handle
         connected.set()
         logger.info('connected %s', handle)
     elif state == zookeeper.CONNECTING_STATE:
         connected.clear()
     elif state == zookeeper.EXPIRED_SESSION_STATE:
         connected.clear()
         if self.handle is not None:
             zookeeper.close(self.handle)
         self.handle = None
         init()
     else:
         logger.critical('unexpected session event %s %s', handle, state)

    def set(self, path, data='', previous_data=None):
        path = '%s%s' % (self.prefix, path)
        try:
            current_data, current_meta = zookeeper.get(self.handle, path, None)
        except zookeeper.NoNodeException:
            if not previous_data:
                try:
                    zookeeper.create(self.handle, path, data, [ZOO_OPEN_ACL_UNSAFE])
                    return True
                except:
                    self.log.exception('Failed to create a missing key %s', path)
                    return False
            else:
                return False
        except:
            self.log.exception('Failed to set key %s', path)
            return False

        version = None
        if previous_data:
            if current_data != previous_data:
                self.log.error('Previous data constraint failed')
                return False
            version = current_meta['version']

        try:
            if version is None:
                zookeeper.set(self.handle, path, data)
            else:
                zookeeper.set(self.handle, path, data, version)
        except:
            self.log.exception('Set failed')
            return False

        return True

 def register_machine(self, machine, data):
     if not isinstance(data, odin_pb2.Machine):
         raise TypeError('data must be a protobuffer Machine')
     zookeeper.create(self._z,
                      machine,
                      data.SerializeToString(),
                      [ZOO_OPEN_ACL_UNSAFE],
                      zookeeper.EPHEMERAL)

  def mark_node_available(self, node_id, data=""):
    """Mark a node available."""

    path = self.AVAILABILITY_NODE + "/" + str(node_id)
    try:
      zookeeper.create(self.handle, path, data, [ZOO_OPEN_ACL_UNSAFE], zookeeper.EPHEMERAL)
      self.logger.info("Node %d: %s is now available" % (node_id, data))
    except zookeeper.NodeExistsException:
      self.logger.warn("%s already exists" % path)

    def write(self, path, contents, ephemeral=False, exclusive=False):
        """ 
        Writes the contents to the path in zookeeper. It will create the path in
        zookeeper if it does not already exist.

        This method will return True if the value is written, False otherwise.
        (The value will not be written if the exclusive is True and the node
        already exists.)
        """
        partial_path = ''

        # We start from the second element because we do not want to inclued
        # the initial empty string before the first "/" because all paths begin
        # with "/". We also don't want to include the final node because that
        # is dealt with later.

        for path_part in path.split("/")[1:-1]:
            partial_path = partial_path + "/" + path_part
            if not(zookeeper.exists(self.handle, partial_path)):
                try:
                    zookeeper.create(self.handle, partial_path, '', [self.acl], 0)
                except zookeeper.NodeExistsException:
                    pass

        exists = zookeeper.exists(self.handle, path)

        # Don't create it if we're exclusive.
        if exists and exclusive:
            return False

        # We make sure that we have the creation flags for ephemeral nodes,
        # otherwise they could be associated with previous connections that
        # have not yet timed out.
        if ephemeral and exists:
            try:
                zookeeper.delete(self.handle, path)
            except zookeeper.NoNodeException:
                pass
            exists = False

        if exists:
            zookeeper.set(self.handle, path, contents)
            return True
        else:
            flags = (ephemeral and zookeeper.EPHEMERAL or 0)
            try:
                zookeeper.create(self.handle, path, contents, [self.acl], flags)
                return True
            except zookeeper.NodeExistsException:
                if not(exclusive):
                    # Woah, something happened between the top and here.
                    # We just restart and retry the whole routine.
                    self.write(path, contents, ephemeral=ephemeral)
                    return True
                else:
                    return False

  def __init__(self, semaphorename, hostname, port, initial_value = 0):
    ZooKeeperBase.__init__(self, hostname, port)

    self.semaphorename = "/" + semaphorename
    try:
      zookeeper.create(self.handle,self.semaphorename,"semaphore top level", [ZOO_OPEN_ACL_UNSAFE],0)
      self.signal(initial_value)
      print "Created new semaphore, OK"
    except zookeeper.NodeExistsException:
      print "Semaphore Already Exists"

  def __init__(self,queuename, hostname, port, is_producer=False):
    ZooKeeperBase.__init__(self, hostname, port)
    self.queuename = "/" + queuename

    if is_producer:
      try:
        zookeeper.create(self.handle,self.queuename,"queue top level", [ZOO_OPEN_ACL_UNSAFE],0)
        print "Created new Queue, OK"
      except zookeeper.NodeExistsException:
        print "Queue Already Exists"

 def test_empty_node(self):
     """
     Test for a bug when instead of empty string we can get
     random data from buffer malloc'ed to hold node contents.
     See ZOOKEEPER-1906 for details
     """
     NODE_PATH = "/zk-python-test-empty-node"
     self.ensureDeleted(NODE_PATH)
     zookeeper.create(self.handle, NODE_PATH, "",
                      [{"perms":0x1f, "scheme":"world", "id" :"anyone"}])
     (data,stat) = zookeeper.get(self.handle, NODE_PATH, None)
     self.assertEqual(data, "", "Data is not empty as expected: " + data)

 def createRootNode(self, root):
     #Create root node recursively from back to front to back
     if root == '/': return
     while True:
         try:
             root = root[:-1]
             zookeeper.create(self.zh, root, '', [ZOO_OPEN_ACL_UNSAFE], 0)
             return
         except zookeeper.NodeExistsException:
             return
         except zookeeper.NoNodeException:
             self.createRootNode(root.rsplit('/', 1)[0] + '/')