class AuthFailedError(ZookeeperError):
    pass class ConfigurationError(KazooException):
    """Raised if the configuration arguments to an object are
    invalid""" class ConnectionClosedError(SessionExpiredError):
    """Connection is closed""" class ConnectionLoss(ZookeeperError):
    pass class KazooException(Exception):
    """Base Kazoo exception that all other kazoo library exceptions
    inherit from""" class NoNodeError(ZookeeperError):
    pass class NodeExistsError(ZookeeperError):
    pass class SessionExpiredError(ZookeeperError):
    pass class WriterNotClosedException(KazooException):
    """Raised if the writer is unable to stop closing when requested.

    .. versionadded:: 1.2
    """ class SequentialThreadingHandler(object):
    """Threading handler for sequentially executing callbacks.

    This handler executes callbacks in a sequential manner. A queue is
    created for each of the callback events, so that each type of event
    has its callback type run sequentially. These are split into two
    queues, one for watch events and one for async result completion
    callbacks.

    Each queue type has a thread worker that pulls the callback event
    off the queue and runs it in the order the client sees it.

    This split helps ensure that watch callbacks won't block session
    re-establishment should the connection be lost during a Zookeeper
    client call.

    Watch and completion callbacks should avoid blocking behavior as
    the next callback of that type won't be run until it completes. If
    you need to block, spawn a new thread and return immediately so
    callbacks can proceed.

    .. note::

        Completion callbacks can block to wait on Zookeeper calls, but
        no other completion callbacks will execute until the callback
        returns.

    """
    name = "sequential_threading_handler"
    timeout_exception = KazooTimeoutError
    sleep_func = staticmethod(time.sleep)
    queue_impl = Queue.Queue
    queue_empty = Queue.Empty

    def __init__(self):
        """Create a :class:`SequentialThreadingHandler` instance"""
        self.callback_queue = self.queue_impl()
        self.completion_queue = self.queue_impl()
        self._running = False
        self._state_change = threading.Lock()
        self._workers = []

    @property
    def running(self):
        return self._running

    def _create_thread_worker(self, queue):
        def _thread_worker():  # pragma: nocover
            while True:
                try:
                    func = queue.get()
                    try:
                        if func is _STOP:
                            break
                        func()
                    except Exception:
                        log.exception("Exception in worker queue thread")
                    finally:
                        queue.task_done()
                        del func  # release before possible idle
                except self.queue_empty:
                    continue

        t = self.spawn(_thread_worker)
        return t

    def start(self):
        """Start the worker threads."""
        with self._state_change:
            if self._running:
                return

            # Spawn our worker threads, we have
            # - A callback worker for watch events to be called
            # - A completion worker for completion events to be called
            for queue in (self.completion_queue, self.callback_queue):
                w = self._create_thread_worker(queue)
                self._workers.append(w)
            self._running = True
            python2atexit.register(self.stop)

    def stop(self):
        """Stop the worker threads and empty all queues."""
        with self._state_change:
            if not self._running:
                return

            self._running = False

            for queue in (self.completion_queue, self.callback_queue):
                queue.put(_STOP)

            self._workers.reverse()
            while self._workers:
                worker = self._workers.pop()
                worker.join()

            # Clear the queues
            self.callback_queue = self.queue_impl()
            self.completion_queue = self.queue_impl()
            python2atexit.unregister(self.stop)

    def select(self, *args, **kwargs):
        return selector_select(*args, **kwargs)

    def socket(self):
        return utils.create_tcp_socket(socket)

    def create_connection(self, *args, **kwargs):
        return utils.create_tcp_connection(socket, *args, **kwargs)

    def create_socket_pair(self):
        return utils.create_socket_pair(socket)

    def event_object(self):
        """Create an appropriate Event object"""
        return threading.Event()

    def lock_object(self):
        """Create a lock object"""
        return threading.Lock()

    def rlock_object(self):
        """Create an appropriate RLock object"""
        return threading.RLock()

    def async_result(self):
        """Create a :class:`AsyncResult` instance"""
        return AsyncResult(self)

    def spawn(self, func, *args, **kwargs):
        t = threading.Thread(target=func, args=args, kwargs=kwargs)
        t.daemon = True
        t.start()
        return t

    def dispatch_callback(self, callback):
        """Dispatch to the callback object

        The callback is put on separate queues to run depending on the
        type as documented for the :class:`SequentialThreadingHandler`.

        """
        self.callback_queue.put(lambda: callback.func(*callback.args)) def capture_exceptions(async_result):
    """Return a new decorated function that propagates the exceptions of the
    wrapped function to an async_result.

    :param async_result: An async result implementing :class:`IAsyncResult`

    """

    def capture(function):
        @functools.wraps(function)
        def captured_function(*args, **kwargs):
            try:
                return function(*args, **kwargs)
            except Exception as exc:
                async_result.set_exception(exc)

        return captured_function

    return capture def wrap(async_result):
    """Return a new decorated function that propagates the return value or
    exception of wrapped function to an async_result.  NOTE: Only propagates a
    non-None return value.

    :param async_result: An async result implementing :class:`IAsyncResult`

    """

    def capture(function):
        @capture_exceptions(async_result)
        def captured_function(*args, **kwargs):
            value = function(*args, **kwargs)
            if value is not None:
                async_result.set(value)
            return value

        return captured_function

    return capture def collect_hosts(hosts):
    """
       Collect a set of hosts and an optional chroot from
       a string or a list of strings.
    """
    if isinstance(hosts, list):
        if hosts[-1].strip().startswith('/'):
            host_ports, chroot = hosts[:-1], hosts[-1]
        else:
            host_ports, chroot = hosts, None
    else:
        host_ports, chroot = hosts.partition("/")[::2]
        host_ports = host_ports.split(",")
        chroot = "/" + chroot if chroot else None

    result = []
    for host_port in host_ports:
        # put all complexity of dealing with
        # IPv4 & IPv6 address:port on the urlsplit
        res = urllib_parse.urlsplit("xxx://" + host_port)
        host = res.hostname
        if host is None:
            raise ValueError("bad hostname")
        port = int(res.port) if res.port else 2181
        result.append((host.strip(), port))

    return result, chroot class ConnectionHandler(object):
    """Zookeeper connection handler"""
    def __init__(self, client, retry_sleeper, logger=None, sasl_options=None):
        self.client = client
        self.handler = client.handler
        self.retry_sleeper = retry_sleeper
        self.logger = logger or log

        # Our event objects
        self.connection_closed = client.handler.event_object()
        self.connection_closed.set()
        self.connection_stopped = client.handler.event_object()
        self.connection_stopped.set()
        self.ping_outstanding = client.handler.event_object()

        self._read_sock = None
        self._write_sock = None

        self._socket = None
        self._xid = None
        self._rw_server = None
        self._ro_mode = False

        self._connection_routine = None

        self.sasl_options = sasl_options
        self.sasl_cli = None

    # This is instance specific to avoid odd thread bug issues in Python
    # during shutdown global cleanup
    @contextmanager
    def _socket_error_handling(self):
        try:
            yield
        except (socket.error, select.error) as e:
            err = getattr(e, 'strerror', e)
            raise ConnectionDropped("socket connection error: %s" % (err,))

    def start(self):
        """Start the connection up"""
        if self.connection_closed.is_set():
            rw_sockets = self.handler.create_socket_pair()
            self._read_sock, self._write_sock = rw_sockets
            self.connection_closed.clear()
        if self._connection_routine:
            raise Exception("Unable to start, connection routine already "
                            "active.")
        self._connection_routine = self.handler.spawn(self.zk_loop)

    def stop(self, timeout=None):
        """Ensure the writer has stopped, wait to see if it does."""
        self.connection_stopped.wait(timeout)
        if self._connection_routine:
            self._connection_routine.join()
            self._connection_routine = None
        return self.connection_stopped.is_set()

    def close(self):
        """Release resources held by the connection

        The connection can be restarted afterwards.
        """
        if not self.connection_stopped.is_set():
            raise Exception("Cannot close connection until it is stopped")
        self.connection_closed.set()
        ws, rs = self._write_sock, self._read_sock
        self._write_sock = self._read_sock = None
        if ws is not None:
            ws.close()
        if rs is not None:
            rs.close()

    def _server_pinger(self):
        """Returns a server pinger iterable, that will ping the next
        server in the list, and apply a back-off between attempts."""
        return RWPinger(self.client.hosts, self.handler.create_connection,
                        self._socket_error_handling)

    def _read_header(self, timeout):
        b = self._read(4, timeout)
        length = int_struct.unpack(b)[0]
        b = self._read(length, timeout)
        header, offset = ReplyHeader.deserialize(b, 0)
        return header, b, offset

    def _read(self, length, timeout):
        msgparts = []
        remaining = length
        with self._socket_error_handling():
            while remaining > 0:
                # Because of SSL framing, a select may not return when using
                # an SSL socket because the underlying physical socket may not
                # have anything to select, but the wrapped object may still
                # have something to read as it has previously gotten enough
                # data from the underlying socket.
                if (hasattr(self._socket, "pending") and
                        self._socket.pending() > 0):
                    pass
                else:
                    s = self.handler.select([self._socket], [], [], timeout)[0]
                    if not s:  # pragma: nocover
                        # If the read list is empty, we got a timeout. We don't
                        # have to check wlist and xlist as we don't set any
                        raise self.handler.timeout_exception(
                            "socket time-out during read")
                try:
                    chunk = self._socket.recv(remaining)
                except ssl.SSLError as e:
                    if e.errno in (ssl.SSL_ERROR_WANT_READ,
                                   ssl.SSL_ERROR_WANT_WRITE):
                        continue
                    else:
                        raise
                if chunk == b'':
                    raise ConnectionDropped('socket connection broken')
                msgparts.append(chunk)
                remaining -= len(chunk)
            return b"".join(msgparts)

    def _invoke(self, timeout, request, xid=None):
        """A special writer used during connection establishment
        only"""
        self._submit(request, timeout, xid)
        zxid = None
        if xid:
            header, buffer, offset = self._read_header(timeout)
            if header.xid != xid:
                raise RuntimeError('xids do not match, expected %r '
                                   'received %r', xid, header.xid)
            if header.zxid > 0:
                zxid = header.zxid
            if header.err:
                callback_exception = EXCEPTIONS[header.err]()
                self.logger.debug(
                    'Received error(xid=%s) %r', xid, callback_exception)
                raise callback_exception
            return zxid

        msg = self._read(4, timeout)
        length = int_struct.unpack(msg)[0]
        msg = self._read(length, timeout)

        if hasattr(request, 'deserialize'):
            try:
                obj, _ = request.deserialize(msg, 0)
            except Exception:
                self.logger.exception(
                    "Exception raised during deserialization "
                    "of request: %s", request)

                # raise ConnectionDropped so connect loop will retry
                raise ConnectionDropped('invalid server response')
            self.logger.log(BLATHER, 'Read response %s', obj)
            return obj, zxid

        return zxid

    def _submit(self, request, timeout, xid=None):
        """Submit a request object with a timeout value and optional
        xid"""
        b = bytearray()
        if xid:
            b.extend(int_struct.pack(xid))
        if request.type:
            b.extend(int_struct.pack(request.type))
        b += request.serialize()
        self.logger.log(
            (BLATHER if isinstance(request, Ping) else logging.DEBUG),
            "Sending request(xid=%s): %s", xid, request)
        self._write(int_struct.pack(len(b)) + b, timeout)

    def _write(self, msg, timeout):
        """Write a raw msg to the socket"""
        sent = 0
        msg_length = len(msg)
        with self._socket_error_handling():
            while sent < msg_length:
                s = self.handler.select([], [self._socket], [], timeout)[1]
                if not s:  # pragma: nocover
                    # If the write list is empty, we got a timeout. We don't
                    # have to check rlist and xlist as we don't set any
                    raise self.handler.timeout_exception("socket time-out"
                                                         " during write")
                msg_slice = buffer(msg, sent)
                try:
                    bytes_sent = self._socket.send(msg_slice)
                except ssl.SSLError as e:
                    if e.errno in (ssl.SSL_ERROR_WANT_READ,
                                   ssl.SSL_ERROR_WANT_WRITE):
                        continue
                    else:
                        raise
                if not bytes_sent:
                    raise ConnectionDropped('socket connection broken')
                sent += bytes_sent

    def _read_watch_event(self, buffer, offset):
        client = self.client
        watch, offset = Watch.deserialize(buffer, offset)
        path = watch.path

        self.logger.debug('Received EVENT: %s', watch)

        watchers = []

        if watch.type in (CREATED_EVENT, CHANGED_EVENT):
            watchers.extend(client._data_watchers.pop(path, []))
        elif watch.type == DELETED_EVENT:
            watchers.extend(client._data_watchers.pop(path, []))
            watchers.extend(client._child_watchers.pop(path, []))
        elif watch.type == CHILD_EVENT:
            watchers.extend(client._child_watchers.pop(path, []))
        else:
            self.logger.warn('Received unknown event %r', watch.type)
            return

        # Strip the chroot if needed
        path = client.unchroot(path)
        ev = WatchedEvent(EVENT_TYPE_MAP[watch.type], client._state, path)

        # Last check to ignore watches if we've been stopped
        if client._stopped.is_set():
            return

        # Dump the watchers to the watch thread
        for watch in watchers:
            client.handler.dispatch_callback(Callback('watch', watch, (ev,)))

    def _read_response(self, header, buffer, offset):
        client = self.client
        request, async_object, xid = client._pending.popleft()
        if header.zxid and header.zxid > 0:
            client.last_zxid = header.zxid
        if header.xid != xid:
            exc = RuntimeError('xids do not match, expected %r '
                               'received %r', xid, header.xid)
            async_object.set_exception(exc)
            raise exc

        # Determine if its an exists request and a no node error
        exists_error = (header.err == NoNodeError.code and
                        request.type == Exists.type)

        # Set the exception if its not an exists error
        if header.err and not exists_error:
            callback_exception = EXCEPTIONS[header.err]()
            self.logger.debug(
                'Received error(xid=%s) %r', xid, callback_exception)
            if async_object:
                async_object.set_exception(callback_exception)
        elif request and async_object:
            if exists_error:
                # It's a NoNodeError, which is fine for an exists
                # request
                async_object.set(None)
            else:
                try:
                    response = request.deserialize(buffer, offset)
                except Exception as exc:
                    self.logger.exception(
                        "Exception raised during deserialization "
                        "of request: %s", request)
                    async_object.set_exception(exc)
                    return
                self.logger.debug(
                    'Received response(xid=%s): %r', xid, response)

                # We special case a Transaction as we have to unchroot things
                if request.type == Transaction.type:
                    response = Transaction.unchroot(client, response)

                async_object.set(response)

            # Determine if watchers should be registered
            watcher = getattr(request, 'watcher', None)
            if not client._stopped.is_set() and watcher:
                if isinstance(request, (GetChildren, GetChildren2)):
                    client._child_watchers[request.path].add(watcher)
                else:
                    client._data_watchers[request.path].add(watcher)

        if isinstance(request, Close):
            self.logger.log(BLATHER, 'Read close response')
            return CLOSE_RESPONSE

    def _read_socket(self, read_timeout):
        """Called when there's something to read on the socket"""
        client = self.client

        header, buffer, offset = self._read_header(read_timeout)
        if header.xid == PING_XID:
            self.logger.log(BLATHER, 'Received Ping')
            self.ping_outstanding.clear()
        elif header.xid == AUTH_XID:
            self.logger.log(BLATHER, 'Received AUTH')

            request, async_object, xid = client._pending.popleft()
            if header.err:
                async_object.set_exception(AuthFailedError())
                client._session_callback(KeeperState.AUTH_FAILED)
            else:
                async_object.set(True)
        elif header.xid == WATCH_XID:
            self._read_watch_event(buffer, offset)
        else:
            self.logger.log(BLATHER, 'Reading for header %r', header)

            return self._read_response(header, buffer, offset)

    def _send_request(self, read_timeout, connect_timeout):
        """Called when we have something to send out on the socket"""
        client = self.client
        try:
            request, async_object = client._queue[0]
        except IndexError:
            # Not actually something on the queue, this can occur if
            # something happens to cancel the request such that we
            # don't clear the socket below after sending
            try:
                # Clear possible inconsistence (no request in the queue
                # but have data in the read socket), which causes cpu to spin.
                self._read_sock.recv(1)
            except OSError:
                pass
            return

        # Special case for testing, if this is a _SessionExpire object
        # then throw a SessionExpiration error as if we were dropped
        if request is _SESSION_EXPIRED:
            raise SessionExpiredError("Session expired: Testing")
        if request is _CONNECTION_DROP:
            raise ConnectionDropped("Connection dropped: Testing")

        # Special case for auth packets
        if request.type == Auth.type:
            xid = AUTH_XID
        else:
            self._xid = (self._xid % 2147483647) + 1
            xid = self._xid

        self._submit(request, connect_timeout, xid)
        client._queue.popleft()
        self._read_sock.recv(1)
        client._pending.append((request, async_object, xid))

    def _send_ping(self, connect_timeout):
        self.ping_outstanding.set()
        self._submit(PingInstance, connect_timeout, PING_XID)

        # Determine if we need to check for a r/w server
        if self._ro_mode:
            result = advance_iterator(self._ro_mode)
            if result:
                self._rw_server = result
                raise RWServerAvailable()

    def zk_loop(self):
        """Main Zookeeper handling loop"""
        self.logger.log(BLATHER, 'ZK loop started')

        self.connection_stopped.clear()

        retry = self.retry_sleeper.copy()
        try:
            while not self.client._stopped.is_set():
                # If the connect_loop returns STOP_CONNECTING, stop retrying
                if retry(self._connect_loop, retry) is STOP_CONNECTING:
                    break
        except RetryFailedError:
            self.logger.warning("Failed connecting to Zookeeper "
                                "within the connection retry policy.")
        finally:
            self.connection_stopped.set()
            self.client._session_callback(KeeperState.CLOSED)
            self.logger.log(BLATHER, 'Connection stopped')

    def _expand_client_hosts(self):
        # Expand the entire list in advance so we can randomize it if needed
        host_ports = []
        for host, port in self.client.hosts:
            try:
                host = host.strip()
                for rhost in socket.getaddrinfo(host, port, 0, 0,
                                                socket.IPPROTO_TCP):
                    host_ports.append((host, rhost[4][0], rhost[4][1]))
            except socket.gaierror as e:
                # Skip hosts that don't resolve
                self.logger.warning("Cannot resolve %s: %s", host, e)
                pass
        if self.client.randomize_hosts:
            random.shuffle(host_ports)
        return host_ports

    def _connect_loop(self, retry):
        # Iterate through the hosts a full cycle before starting over
        status = None
        host_ports = self._expand_client_hosts()

        # Check for an empty hostlist, indicating none resolved
        if len(host_ports) == 0:
            raise ForceRetryError('No host resolved. Reconnecting')

        for host, hostip, port in host_ports:
            if self.client._stopped.is_set():
                status = STOP_CONNECTING
                break
            status = self._connect_attempt(host, hostip, port, retry)
            if status is STOP_CONNECTING:
                break

        if status is STOP_CONNECTING:
            return STOP_CONNECTING
        else:
            raise ForceRetryError('Reconnecting')

    def _connect_attempt(self, host, hostip, port, retry):
        client = self.client
        KazooTimeoutError = self.handler.timeout_exception

        self._socket = None

        # Were we given a r/w server? If so, use that instead
        if self._rw_server:
            self.logger.log(BLATHER,
                            "Found r/w server to use, %s:%s", host, port)
            host, port = self._rw_server
            self._rw_server = None

        if client._state != KeeperState.CONNECTING:
            client._session_callback(KeeperState.CONNECTING)

        try:
            self._xid = 0
            read_timeout, connect_timeout = self._connect(host, hostip, port)
            read_timeout = read_timeout / 1000.0
            connect_timeout = connect_timeout / 1000.0
            retry.reset()
            self.ping_outstanding.clear()
            last_send = time.time()
            with self._socket_error_handling():
                while True:
                    # Watch for something to read or send
                    jitter_time = random.randint(1, 40) / 100.0
                    deadline = last_send + read_timeout / 2.0 - jitter_time
                    # Ensure our timeout is positive
                    timeout = max([deadline - time.time(), jitter_time])
                    s = self.handler.select([self._socket, self._read_sock],
                                            [], [], timeout)[0]

                    if not s:
                        if self.ping_outstanding.is_set():
                            self.ping_outstanding.clear()
                            raise ConnectionDropped(
                                "outstanding heartbeat ping not received")
                    else:
                        if self._socket in s:
                            response = self._read_socket(read_timeout)
                            if response == CLOSE_RESPONSE:
                                break
                        # Check if any requests need sending before proceeding
                        # to process more responses.  Otherwise the responses
                        # may choke out the requests.  See PR#633.
                        if self._read_sock in s:
                            self._send_request(read_timeout, connect_timeout)
                            # Requests act as implicit pings.
                            last_send = time.time()
                            continue

                    if time.time() >= deadline:
                        self._send_ping(connect_timeout)
                        last_send = time.time()
            self.logger.info('Closing connection to %s:%s', host, port)
            client._session_callback(KeeperState.CLOSED)
            return STOP_CONNECTING
        except (ConnectionDropped, KazooTimeoutError) as e:
            if isinstance(e, ConnectionDropped):
                self.logger.warning('Connection dropped: %s', e)
            else:
                self.logger.warning('Connection time-out: %s', e)
            if client._state != KeeperState.CONNECTING:
                self.logger.warning("Transition to CONNECTING")
                client._session_callback(KeeperState.CONNECTING)
        except AuthFailedError as err:
            retry.reset()
            self.logger.warning('AUTH_FAILED closing: %s', err)
            client._session_callback(KeeperState.AUTH_FAILED)
            return STOP_CONNECTING
        except SessionExpiredError:
            retry.reset()
            self.logger.warning('Session has expired')
            client._session_callback(KeeperState.EXPIRED_SESSION)
        except RWServerAvailable:
            retry.reset()
            self.logger.warning('Found a RW server, dropping connection')
            client._session_callback(KeeperState.CONNECTING)
        except Exception:
            self.logger.exception('Unhandled exception in connection loop')
            raise
        finally:
            if self._socket is not None:
                self._socket.close()

    def _connect(self, host, hostip, port):
        client = self.client
        self.logger.info('Connecting to %s(%s):%s, use_ssl: %r',
                         host, hostip, port, self.client.use_ssl)

        self.logger.log(BLATHER,
                        '    Using session_id: %r session_passwd: %s',
                        client._session_id,
                        hexlify(client._session_passwd))

        with self._socket_error_handling():
            self._socket = self.handler.create_connection(
                address=(hostip, port),
                timeout=client._session_timeout / 1000.0,
                use_ssl=self.client.use_ssl,
                keyfile=self.client.keyfile,
                certfile=self.client.certfile,
                ca=self.client.ca,
                keyfile_password=self.client.keyfile_password,
                verify_certs=self.client.verify_certs,
            )

        self._socket.setblocking(0)

        connect = Connect(0, client.last_zxid, client._session_timeout,
                          client._session_id or 0, client._session_passwd,
                          client.read_only)

        connect_result, zxid = self._invoke(
            client._session_timeout / 1000.0 / len(client.hosts), connect)

        if connect_result.time_out <= 0:
            raise SessionExpiredError("Session has expired")

        if zxid:
            client.last_zxid = zxid

        # Load return values
        client._session_id = connect_result.session_id
        client._protocol_version = connect_result.protocol_version
        negotiated_session_timeout = connect_result.time_out
        connect_timeout = negotiated_session_timeout / len(client.hosts)
        read_timeout = negotiated_session_timeout * 2.0 / 3.0
        client._session_passwd = connect_result.passwd

        self.logger.log(BLATHER,
                        'Session created, session_id: %r session_passwd: %s\n'
                        '    negotiated session timeout: %s\n'
                        '    connect timeout: %s\n'
                        '    read timeout: %s', client._session_id,
                        hexlify(client._session_passwd),
                        negotiated_session_timeout, connect_timeout,
                        read_timeout)

        if connect_result.read_only:
            client._session_callback(KeeperState.CONNECTED_RO)
            self._ro_mode = iter(self._server_pinger())
        else:
            client._session_callback(KeeperState.CONNECTED)
            self._ro_mode = None

        if self.sasl_options is not None:
            self._authenticate_with_sasl(host, connect_timeout / 1000.0)

        # Get a copy of the auth data before iterating, in case it is
        # changed.
        client_auth_data_copy = copy.copy(client.auth_data)

        for scheme, auth in client_auth_data_copy:
            ap = Auth(0, scheme, auth)
            zxid = self._invoke(connect_timeout / 1000.0, ap, xid=AUTH_XID)
            if zxid:
                client.last_zxid = zxid

        return read_timeout, connect_timeout

    def _authenticate_with_sasl(self, host, timeout):
        """Establish a SASL authenticated connection to the server.
        """
        if not PURESASL_AVAILABLE:
            raise SASLException('Missing SASL support')

        if 'service' not in self.sasl_options:
            self.sasl_options['service'] = 'zookeeper'

        # NOTE: Zookeeper hardcoded the domain for Digest authentication
        # instead of using the hostname. See
        # zookeeper/util/SecurityUtils.java#L74 and Server/Client
        # initializations.
        if self.sasl_options['mechanism'] == 'DIGEST-MD5':
            host = 'zk-sasl-md5'

        sasl_cli = self.client.sasl_cli = puresasl.client.SASLClient(
            host=host,
            **self.sasl_options
        )

        # Inititalize the process with an empty challenge token
        challenge = None
        xid = 0

        while True:
            if sasl_cli.complete:
                break

            try:
                response = sasl_cli.process(challenge=challenge)
            except puresasl.SASLError as err:
                six.reraise(
                    SASLException,
                    SASLException('library error: %s' % err.message),
                    sys.exc_info()[2]
                )
            except puresasl.SASLProtocolException as err:
                six.reraise(
                    AuthFailedError,
                    AuthFailedError('protocol error: %s' % err.message),
                    sys.exc_info()[2]
                )
            except Exception as err:
                six.reraise(
                    AuthFailedError,
                    AuthFailedError('Unknown error: %s' % err),
                    sys.exc_info()[2]
                )

            if sasl_cli.complete and not response:
                break
            elif response is None:
                response = b''

            xid = (xid % 2147483647) + 1

            request = SASL(response)
            self._submit(request, timeout, xid)

            try:
                header, buffer, offset = self._read_header(timeout)
            except ConnectionDropped:
                # Zookeeper simply drops connections with failed authentication
                six.reraise(
                    AuthFailedError,
                    AuthFailedError('Connection dropped in SASL'),
                    sys.exc_info()[2]
                )

            if header.xid != xid:
                raise RuntimeError('xids do not match, expected %r '
                                   'received %r', xid, header.xid)

            if header.zxid > 0:
                self.client.last_zxid = header.zxid

            if header.err:
                callback_exception = EXCEPTIONS[header.err]()
                self.logger.debug(
                    'Received error(xid=%s) %r', xid, callback_exception)
                raise callback_exception

            challenge, _ = SASL.deserialize(buffer, offset)

        # If we made it here, authentication is ok, and we are connected.
        # Remove sensible information from the object.
        sasl_cli.dispose() def read_string(buffer, offset):
def read_acl(bytes, offset):
def write_string(bytes):
def write_buffer(bytes):
def read_buffer(bytes, offset):
    def serialize(cls):
    def serialize(cls):
    def serialize(self):
    def deserialize(cls, bytes, offset):
    def serialize(self):
    def deserialize(cls, bytes, offset):
    def serialize(self):
    def deserialize(self, bytes, offset):
    def serialize(self):
    def deserialize(cls, bytes, offset):
    def serialize(self):
    def deserialize(cls, bytes, offset):
    def serialize(self):
    def deserialize(cls, bytes, offset):
    def serialize(self):
    def deserialize(cls, bytes, offset):
    def serialize(self):
    def deserialize(cls, bytes, offset):
    def serialize(self):
    def deserialize(cls, bytes, offset):
    def serialize(self):
    def deserialize(cls, buffer, offset):
    def serialize(self):
    def deserialize(cls, bytes, offset):
    def serialize(self):
    def serialize(self):
    def deserialize(cls, bytes, offset):
    def unchroot(client, response):
    def serialize(self):
    def deserialize(cls, bytes, offset):
    def serialize(self):
    def deserialize(cls, bytes, offset):
    def serialize(self):
    def serialize(self):
    def deserialize(cls, bytes, offset):
    def deserialize(cls, bytes, offset):
    def deserialize(cls, bytes, offset):
    def serialize(self):
    def deserialize(cls, bytes, offset):
class Close(namedtuple('Close', '')):
class Ping(namedtuple('Ping', '')):
class Connect(namedtuple('Connect', 'protocol_version last_zxid_seen'
                         ' time_out session_id passwd read_only')):
class Create(namedtuple('Create', 'path data acl flags')):
class Delete(namedtuple('Delete', 'path version')):
class Exists(namedtuple('Exists', 'path watcher')):
class GetData(namedtuple('GetData', 'path watcher')):
class SetData(namedtuple('SetData', 'path data version')):
class GetACL(namedtuple('GetACL', 'path')):
class SetACL(namedtuple('SetACL', 'path acls version')):
class GetChildren(namedtuple('GetChildren', 'path watcher')):
class Sync(namedtuple('Sync', 'path')):
class GetChildren2(namedtuple('GetChildren2', 'path watcher')):
class CheckVersion(namedtuple('CheckVersion', 'path version')):
class Transaction(namedtuple('Transaction', 'operations')):
class Create2(namedtuple('Create2', 'path data acl flags')):
class Reconfig(namedtuple('Reconfig',
                          'joining leaving new_members config_id')):
class Auth(namedtuple('Auth', 'auth_type scheme auth')):
class SASL(namedtuple('SASL', 'challenge')):
class Watch(namedtuple('Watch', 'type state path')):
class ReplyHeader(namedtuple('ReplyHeader', 'xid, zxid, err')):
class MultiHeader(namedtuple('MultiHeader', 'type done err')): class Callback(namedtuple('Callback', ('type', 'func', 'args'))):
    """A callback that is handed to a handler for dispatch

    :param type: Type of the callback, currently is only 'watch'
    :param func: Callback function
    :param args: Argument list for the callback function

    """ class EventType(object):
    """Zookeeper Event

    Represents a Zookeeper event. Events trigger watch functions which
    will receive a :class:`EventType` attribute as their event
    argument.

    .. attribute:: CREATED

        A node has been created.

    .. attribute:: DELETED

        A node has been deleted.

    .. attribute:: CHANGED

        The data for a node has changed.

    .. attribute:: CHILD

        The children under a node have changed (a child was added or
        removed). This event does not indicate the data for a child
        node has changed, which must have its own watch established.

    .. attribute:: NONE

        The connection state has been altered.

    """
    CREATED = 'CREATED'
    DELETED = 'DELETED'
    CHANGED = 'CHANGED'
    CHILD = 'CHILD'
    NONE = 'NONE' class KazooState(object):
    """High level connection state values

    States inspired by Netflix Curator.

    .. attribute:: SUSPENDED

        The connection has been lost but may be recovered. We should
        operate in a "safe mode" until then. When the connection is
        resumed, it may be discovered that the session expired. A
        client should not assume that locks are valid during this
        time.

    .. attribute:: CONNECTED

        The connection is alive and well.

    .. attribute:: LOST

        The connection has been confirmed dead. Any ephemeral nodes
        will need to be recreated upon re-establishing a connection.
        If locks were acquired or recipes using ephemeral nodes are in
        use, they can be considered lost as well.

    """
    SUSPENDED = "SUSPENDED"
    CONNECTED = "CONNECTED"
    LOST = "LOST" class KeeperState(object):
    """Zookeeper State

    Represents the Zookeeper state. Watch functions will receive a
    :class:`KeeperState` attribute as their state argument.

    .. attribute:: AUTH_FAILED

        Authentication has failed, this is an unrecoverable error.

    .. attribute:: CONNECTED

        Zookeeper is connected.

    .. attribute:: CONNECTED_RO

        Zookeeper is connected in read-only state.

    .. attribute:: CONNECTING

        Zookeeper is currently attempting to establish a connection.

    .. attribute:: EXPIRED_SESSION

        The prior session was invalid, all prior ephemeral nodes are
        gone.

    """
    AUTH_FAILED = 'AUTH_FAILED'
    CONNECTED = 'CONNECTED'
    CONNECTED_RO = 'CONNECTED_RO'
    CONNECTING = 'CONNECTING'
    CLOSED = 'CLOSED'
    EXPIRED_SESSION = 'EXPIRED_SESSION' class WatchedEvent(namedtuple('WatchedEvent', ('type', 'state', 'path'))):
    """A change on ZooKeeper that a Watcher is able to respond to.

    The :class:`WatchedEvent` includes exactly what happened, the
    current state of ZooKeeper, and the path of the node that was
    involved in the event. An instance of :class:`WatchedEvent` will be
    passed to registered watch functions.

    .. attribute:: type

        A :class:`EventType` attribute indicating the event type.

    .. attribute:: state

        A :class:`KeeperState` attribute indicating the Zookeeper
        state.

    .. attribute:: path

        The path of the node for the watch event.

    """ class KazooRetry(object):
    """Helper for retrying a method in the face of retry-able
    exceptions"""
    RETRY_EXCEPTIONS = (
        ConnectionLoss,
        OperationTimeoutError,
        ForceRetryError
    )

    EXPIRED_EXCEPTIONS = (
        SessionExpiredError,
    )

    def __init__(self, max_tries=1, delay=0.1, backoff=2, max_jitter=0.4,
                 max_delay=60.0, ignore_expire=True, sleep_func=time.sleep,
                 deadline=None, interrupt=None):
        """Create a :class:`KazooRetry` instance for retrying function
        calls.

        :param max_tries: How many times to retry the command. -1 means
                          infinite tries.
        :param delay: Initial delay between retry attempts.
        :param backoff: Backoff multiplier between retry attempts.
                        Defaults to 2 for exponential backoff.
        :param max_jitter: Percentage of jitter to apply to each retry's delay
                           to ensure all clients to do not hammer the server
                           at the same time. Between 0.0 and 1.0.
        :param max_delay: Maximum delay in seconds, regardless of other
                          backoff settings. Defaults to one minute.
        :param ignore_expire:
            Whether a session expiration should be ignored and treated
            as a retry-able command.
        :param interrupt:
            Function that will be called with no args that may return
            True if the retry should be ceased immediately. This will
            be called no more than every 0.1 seconds during a wait
            between retries.

        """
        self.max_tries = max_tries
        self.delay = delay
        self.backoff = backoff
        # Ensure max_jitter is in (0, 1)
        self.max_jitter = max(min(max_jitter, 1.0), 0.0)
        self.max_delay = float(max_delay)
        self._attempts = 0
        self._cur_delay = delay
        self.deadline = deadline
        self._cur_stoptime = None
        self.sleep_func = sleep_func
        self.retry_exceptions = self.RETRY_EXCEPTIONS
        self.interrupt = interrupt
        if ignore_expire:
            self.retry_exceptions += self.EXPIRED_EXCEPTIONS

    def reset(self):
        """Reset the attempt counter"""
        self._attempts = 0
        self._cur_delay = self.delay
        self._cur_stoptime = None

    def copy(self):
        """Return a clone of this retry manager"""
        obj = KazooRetry(max_tries=self.max_tries,
                         delay=self.delay,
                         backoff=self.backoff,
                         max_jitter=self.max_jitter,
                         max_delay=self.max_delay,
                         sleep_func=self.sleep_func,
                         deadline=self.deadline,
                         interrupt=self.interrupt)
        obj.retry_exceptions = self.retry_exceptions
        return obj

    def __call__(self, func, *args, **kwargs):
        """Call a function with arguments until it completes without
        throwing a Kazoo exception

        :param func: Function to call
        :param args: Positional arguments to call the function with
        :params kwargs: Keyword arguments to call the function with

        The function will be called until it doesn't throw one of the
        retryable exceptions (ConnectionLoss, OperationTimeout, or
        ForceRetryError), and optionally retrying on session
        expiration.

        """
        self.reset()

        while True:
            try:
                if self.deadline is not None and self._cur_stoptime is None:
                    self._cur_stoptime = time.time() + self.deadline
                return func(*args, **kwargs)
            except ConnectionClosedError:
                raise
            except self.retry_exceptions:
                # Note: max_tries == -1 means infinite tries.
                if self._attempts == self.max_tries:
                    raise RetryFailedError("Too many retry attempts")
                self._attempts += 1
                jitter = random.uniform(1.0-self.max_jitter,
                                        1.0+self.max_jitter)
                sleeptime = self._cur_delay * jitter

                if self._cur_stoptime is not None and \
                   time.time() + sleeptime >= self._cur_stoptime:
                    raise RetryFailedError("Exceeded retry deadline")

                if self.interrupt:
                    remain_time = sleeptime
                    while remain_time > 0:
                        # Break the time period down and sleep for no
                        # longer than 0.1 before calling the interrupt
                        self.sleep_func(min(0.1, remain_time))
                        remain_time -= 0.1
                        if self.interrupt():
                            raise InterruptedError()
                else:
                    self.sleep_func(sleeptime)
                self._cur_delay = min(sleeptime * self.backoff,
                                      self.max_delay) class ACL(namedtuple('ACL', 'perms id')):
    """An ACL for a Zookeeper Node

    An ACL object is created by using an :class:`Id` object along with
    a :class:`Permissions` setting. For convenience,
    :meth:`make_digest_acl` should be used to create an ACL object with
    the desired scheme, id, and permissions.
    """
    @property
    def acl_list(self):
        perms = []
        if self.perms & Permissions.ALL == Permissions.ALL:
            perms.append('ALL')
            return perms
        if self.perms & Permissions.READ == Permissions.READ:
            perms.append('READ')
        if self.perms & Permissions.WRITE == Permissions.WRITE:
            perms.append('WRITE')
        if self.perms & Permissions.CREATE == Permissions.CREATE:
            perms.append('CREATE')
        if self.perms & Permissions.DELETE == Permissions.DELETE:
            perms.append('DELETE')
        if self.perms & Permissions.ADMIN == Permissions.ADMIN:
            perms.append('ADMIN')
        return perms

    def __repr__(self):
        return 'ACL(perms=%r, acl_list=%s, id=%r)' % (
            self.perms, self.acl_list, self.id) OPEN_ACL_UNSAFE = [ACL(Permissions.ALL, ANYONE_ID_UNSAFE)] class Barrier(object):
    """Kazoo Barrier

    Implements a barrier to block processing of a set of nodes until
    a condition is met at which point the nodes will be allowed to
    proceed. The barrier is in place if its node exists.

    .. warning::

        The :meth:`wait` function does not handle connection loss and
        may raise :exc:`~kazoo.exceptions.ConnectionLossException` if
        the connection is lost while waiting.

    """
    def __init__(self, client, path):
        """Create a Kazoo Barrier

        :param client: A :class:`~kazoo.client.KazooClient` instance.
        :param path: The barrier path to use.

        """
        self.client = client
        self.path = path

    def create(self):
        """Establish the barrier if it doesn't exist already"""
        self.client.retry(self.client.ensure_path, self.path)

    def remove(self):
        """Remove the barrier

        :returns: Whether the barrier actually needed to be removed.
        :rtype: bool

        """
        try:
            self.client.retry(self.client.delete, self.path)
            return True
        except NoNodeError:
            return False

    def wait(self, timeout=None):
        """Wait on the barrier to be cleared

        :returns: True if the barrier has been cleared, otherwise
                  False.
        :rtype: bool

        """
        cleared = self.client.handler.event_object()

        def wait_for_clear(event):
            if event.type == EventType.DELETED:
                cleared.set()

        exists = self.client.exists(self.path, watch=wait_for_clear)
        if not exists:
            return True

        cleared.wait(timeout)
        return cleared.is_set() class DoubleBarrier(object):
    """Kazoo Double Barrier

    Double barriers are used to synchronize the beginning and end of
    a distributed task. The barrier blocks when entering it until all
    the members have joined, and blocks when leaving until all the
    members have left.

    .. note::

        You should register a listener for session loss as the process
        will no longer be part of the barrier once the session is
        gone. Connection losses will be retried with the default retry
        policy.

    """
    def __init__(self, client, path, num_clients, identifier=None):
        """Create a Double Barrier

        :param client: A :class:`~kazoo.client.KazooClient` instance.
        :param path: The barrier path to use.
        :param num_clients: How many clients must enter the barrier to
                            proceed.
        :type num_clients: int
        :param identifier: An identifier to use for this member of the
                           barrier when participating. Defaults to the
                           hostname + process id.

        """
        self.client = client
        self.path = path
        self.num_clients = num_clients
        self._identifier = identifier or '%s-%s' % (
            socket.getfqdn(), os.getpid())
        self.participating = False
        self.assured_path = False
        self.node_name = uuid.uuid4().hex
        self.create_path = self.path + "/" + self.node_name

    def enter(self):
        """Enter the barrier, blocks until all nodes have entered"""
        try:
            self.client.retry(self._inner_enter)
            self.participating = True
        except KazooException:
            # We failed to enter, best effort cleanup
            self._best_effort_cleanup()
            self.participating = False

    def _inner_enter(self):
        # make sure our barrier parent node exists
        if not self.assured_path:
            self.client.ensure_path(self.path)
            self.assured_path = True

        ready = self.client.handler.event_object()

        try:
            self.client.create(
                self.create_path,
                self._identifier.encode('utf-8'), ephemeral=True)
        except NodeExistsError:
            pass

        def created(event):
            if event.type == EventType.CREATED:
                ready.set()

        self.client.exists(self.path + '/' + 'ready', watch=created)

        children = self.client.get_children(self.path)

        if len(children) < self.num_clients:
            ready.wait()
        else:
            self.client.ensure_path(self.path + '/ready')
        return True

    def leave(self):
        """Leave the barrier, blocks until all nodes have left"""
        try:
            self.client.retry(self._inner_leave)
        except KazooException:  # pragma: nocover
            # Failed to cleanly leave
            self._best_effort_cleanup()
        self.participating = False

    def _inner_leave(self):
        # Delete the ready node if its around
        try:
            self.client.delete(self.path + '/ready')
        except NoNodeError:
            pass

        while True:
            children = self.client.get_children(self.path)
            if not children:
                return True

            if len(children) == 1 and children[0] == self.node_name:
                self.client.delete(self.create_path)
                return True

            children.sort()

            ready = self.client.handler.event_object()

            def deleted(event):
                if event.type == EventType.DELETED:
                    ready.set()

            if self.node_name == children[0]:
                # We're first, wait on the highest to leave
                if not self.client.exists(self.path + '/' + children[-1],
                                          watch=deleted):
                    continue

                ready.wait()
                continue

            # Delete our node
            self.client.delete(self.create_path)

            # Wait on the first
            if not self.client.exists(self.path + '/' + children[0],
                                      watch=deleted):
                continue

            # Wait for the lowest to be deleted
            ready.wait()

    def _best_effort_cleanup(self):
        try:
            self.client.retry(self.client.delete, self.create_path)
        except NoNodeError:
            pass class Counter(object):
    """Kazoo Counter

    A shared counter of either int or float values. Changes to the
    counter are done atomically. The general retry policy is used to
    retry operations if concurrent changes are detected.

    The data is marshaled using `repr(value)` and converted back using
    `type(counter.default)(value)` both using an ascii encoding. As
    such other data types might be used for the counter value.

    If you would like to support clients updating the same znode path using
    either kazoo's counter recipe or curator's SharedCount recipe, you will
    need to enable the support_curator flag. This flag limits
    support to integers only and does not use ascii encoding as described
    above.

    Counter changes can raise
    :class:`~kazoo.exceptions.BadVersionError` if the retry policy
    wasn't able to apply a change.

    Example usage:

    .. code-block:: python

        zk = KazooClient()
        zk.start()
        counter = zk.Counter("/int")
        counter += 2
        counter -= 1
        counter.value == 1
        counter.pre_value == 2
        counter.post_value == 1

        counter = zk.Counter("/float", default=1.0)
        counter += 2.0
        counter.value == 3.0
        counter.pre_value == 1.0
        counter.post_value == 3.0

        counter = zk.Counter("/curator", support_curator=True)
        counter += 2
        counter -= 1
        counter.value == 1
        counter.pre_value == 2
        counter.post_value == 1

    """
    def __init__(self, client, path, default=0, support_curator=False):
        """Create a Kazoo Counter

        :param client: A :class:`~kazoo.client.KazooClient` instance.
        :param path: The counter path to use.
        :param default: The default value to use for new counter paths.
        :param support_curator: Enable if support for curator's SharedCount
                                recipe is desired.

        """
        self.client = client
        self.path = path
        self.default = default
        self.default_type = type(default)
        self.support_curator = support_curator
        self._ensured_path = False
        self.pre_value = None
        self.post_value = None
        if self.support_curator and not isinstance(self.default, int):
            raise TypeError("when support_curator is enabled the default "
                            "type must be an int")

    def _ensure_node(self):
        if not self._ensured_path:
            # make sure our node exists
            self.client.ensure_path(self.path)
            self._ensured_path = True

    def _value(self):
        self._ensure_node()
        old, stat = self.client.get(self.path)
        if self.support_curator:
            old = struct.unpack(">i", old)[0] if old != b'' else self.default
        else:
            old = old.decode('ascii') if old != b'' else self.default
        version = stat.version
        data = self.default_type(old)
        return data, version

    @property
    def value(self):
        return self._value()[0]

    def _change(self, value):
        if not isinstance(value, self.default_type):
            raise TypeError('invalid type for value change')
        self.client.retry(self._inner_change, value)
        return self

    def _inner_change(self, value):
        self.pre_value, version = self._value()
        post_value = self.pre_value + value
        if self.support_curator:
            data = struct.pack(">i", post_value)
        else:
            data = repr(post_value).encode('ascii')
        try:
            self.client.set(self.path, data, version=version)
        except BadVersionError:  # pragma: nocover
            self.post_value = None
            raise ForceRetryError()
        self.post_value = post_value

    def __add__(self, value):
        """Add value to counter."""
        return self._change(value)

    def __sub__(self, value):
        """Subtract value from counter."""
        return self._change(-value) class NonBlockingLease(object):
    """Exclusive lease that does not block.

    An exclusive lease ensures that only one client at a time owns the lease.
    The client may renew the lease without losing it by obtaining a new lease
    with the same path and same identity.  The lease object evaluates to True
    if the lease was obtained.

    A common use case is a situation where a task should only run on a single
    host.  In this case, the clients that did not obtain the lease should exit
    without performing the protected task.

    The lease stores time stamps using client clocks, and will therefore only
    work if client clocks are roughly synchronised.  It uses UTC, and works
    across time zones and daylight savings.

    Example usage: with a :class:`~kazoo.client.KazooClient` instance::

        zk = KazooClient()
        zk.start()
        # Hold lease over an hour in order to keep job on same machine,
        # with failover if it dies.
        lease = zk.NonBlockingLease(
            "/db_leases/hourly_cleanup", datetime.timedelta(minutes = 70),
            identifier = "DB hourly cleanup on " + socket.gethostname())
        if lease:
            do_hourly_database_cleanup()
    """

    # Bump when storage format changes
    _version = 1
    _date_format = "%Y-%m-%dT%H:%M:%S"
    _byte_encoding = 'utf-8'

    def __init__(self, client, path, duration, identifier=None,
                 utcnow=datetime.datetime.utcnow):
        """Create a non-blocking lease.

        :param client: A :class:`~kazoo.client.KazooClient` instance.
        :param path: The lease path to use.
        :param duration: Duration during which the lease is reserved.  A
                         :class:`~datetime.timedelta` instance.
        :param identifier: Unique name to use for this lease holder. Reuse in
                           order to renew the lease. Defaults to
                           :meth:`socket.gethostname()`.
        :param utcnow: Clock function, by default returning
                       :meth:`datetime.datetime.utcnow()`. Used for testing.

        """
        ident = identifier or socket.gethostname()
        self.obtained = False
        self._attempt_obtaining(client, path, duration, ident, utcnow)

    def _attempt_obtaining(self, client, path, duration, ident, utcnow):
        client.ensure_path(path)
        holder_path = path + "/lease_holder"
        lock = client.Lock(path, ident)
        try:
            with lock:
                now = utcnow()
                if client.exists(holder_path):
                    raw, _ = client.get(holder_path)
                    data = self._decode(raw)
                    if data["version"] != self._version:
                        # We need an upgrade, let someone else take the lease
                        return
                    current_end = datetime.datetime.strptime(data['end'],
                                                             self._date_format)
                    if data['holder'] != ident and now < current_end:
                        # Another client is still holding the lease
                        return
                    client.delete(holder_path)
                end_lease = (now + duration).strftime(self._date_format)
                new_data = {'version': self._version, 'holder': ident,
                            'end': end_lease}
                client.create(holder_path, self._encode(new_data))
                self.obtained = True

        except CancelledError:
            pass

    def _encode(self, data_dict):
        return json.dumps(data_dict).encode(self._byte_encoding)

    def _decode(self, raw):
        return json.loads(raw.decode(self._byte_encoding))

    # Python 2.x
    def __nonzero__(self):
        return self.obtained

    # Python 3.x
    def __bool__(self):
        return self.obtained class MultiNonBlockingLease(object):
    """Exclusive lease for multiple clients.

    This type of lease is useful when a limited set of hosts should run a
    particular task. It will attempt to obtain leases trying a sequence of
    ZooKeeper lease paths.

    :param client: A :class:`~kazoo.client.KazooClient` instance.
    :param count: Number of host leases allowed.
    :param path: ZooKeeper path under which lease files are stored.
    :param duration: Duration during which the lease is reserved.  A
                     :class:`~datetime.timedelta` instance.
    :param identifier: Unique name to use for this lease holder. Reuse in order
                       to renew the lease.
           Defaults do :meth:`socket.gethostname()`.
    :param utcnow: Clock function, by default returning
                   :meth:`datetime.datetime.utcnow()`.  Used for testing.

    """

    def __init__(self, client, count, path, duration, identifier=None,
                 utcnow=datetime.datetime.utcnow):
        self.obtained = False
        for num in range(count):
            ls = NonBlockingLease(client, '%s/%d' % (path, num), duration,
                                  identifier=identifier, utcnow=utcnow)
            if ls:
                self.obtained = True
                break

    # Python 2.x
    def __nonzero__(self):
        return self.obtained

    # Python 3.x
    def __bool__(self):
        return self.obtained class Lock(object):
    """Kazoo Lock

    Example usage with a :class:`~kazoo.client.KazooClient` instance:

    .. code-block:: python

        zk = KazooClient()
        zk.start()
        lock = zk.Lock("/lockpath", "my-identifier")
        with lock:  # blocks waiting for lock acquisition
            # do something with the lock

    Note: This lock is not *re-entrant*. Repeated calls after already
    acquired will block.

    This is an exclusive lock. For a read/write lock, see :class:`WriteLock`
    and :class:`ReadLock`.

    """

    # Node name, after the contender UUID, before the sequence
    # number. Involved in read/write locks.
    _NODE_NAME = "__lock__"

    # Node names which exclude this contender when present at a lower
    # sequence number. Involved in read/write locks.
    _EXCLUDE_NAMES = ["__lock__"]

    def __init__(self, client, path, identifier=None, extra_lock_patterns=()):
        """Create a Kazoo lock.

        :param client: A :class:`~kazoo.client.KazooClient` instance.
        :param path: The lock path to use.
        :param identifier: Name to use for this lock contender. This can be
                          useful for querying to see who the current lock
                          contenders are.
        :param extra_lock_patterns: Strings that will be used to
                                    identify other znode in the path
                                    that should be considered contenders
                                    for this lock.
                                    Use this for cross-implementation
                                    compatibility.

        .. versionadded:: 2.7.1
            The extra_lock_patterns option.
        """
        self.client = client
        self.path = path
        self._exclude_names = set(
            self._EXCLUDE_NAMES + list(extra_lock_patterns)
        )
        self._contenders_re = re.compile(
            r"(?:{patterns})(-?\d{{10}})$".format(
                patterns="|".join(self._exclude_names)
            )
        )

        # some data is written to the node. this can be queried via
        # contenders() to see who is contending for the lock
        self.data = str(identifier or "").encode("utf-8")
        self.node = None

        self.wake_event = client.handler.event_object()

        # props to Netflix Curator for this trick. It is possible for our
        # create request to succeed on the server, but for a failure to
        # prevent us from getting back the full path name. We prefix our
        # lock name with a uuid and can check for its presence on retry.
        self.prefix = uuid.uuid4().hex + self._NODE_NAME
        self.create_path = self.path + "/" + self.prefix

        self.create_tried = False
        self.is_acquired = False
        self.assured_path = False
        self.cancelled = False
        self._retry = KazooRetry(
            max_tries=None, sleep_func=client.handler.sleep_func
        )
        self._lock = client.handler.lock_object()

    def _ensure_path(self):
        self.client.ensure_path(self.path)
        self.assured_path = True

    def cancel(self):
        """Cancel a pending lock acquire."""
        self.cancelled = True
        self.wake_event.set()

    def acquire(self, blocking=True, timeout=None, ephemeral=True):
        """
        Acquire the lock. By defaults blocks and waits forever.

        :param blocking: Block until lock is obtained or return immediately.
        :type blocking: bool
        :param timeout: Don't wait forever to acquire the lock.
        :type timeout: float or None
        :param ephemeral: Don't use ephemeral znode for the lock.
        :type ephemeral: bool

        :returns: Was the lock acquired?
        :rtype: bool

        :raises: :exc:`~kazoo.exceptions.LockTimeout` if the lock
                 wasn't acquired within `timeout` seconds.

        .. warning::

            When :attr:`ephemeral` is set to False session expiration
            will not release the lock and must be handled separately.

        .. versionadded:: 1.1
            The timeout option.

        .. versionadded:: 2.4.1
            The ephemeral option.
        """

        def _acquire_lock():
            got_it = self._lock.acquire(False)
            if not got_it:
                raise ForceRetryError()
            return True

        retry = self._retry.copy()
        retry.deadline = timeout

        # Ensure we are locked so that we avoid multiple threads in
        # this acquistion routine at the same time...
        locked = self._lock.acquire(False)
        if not locked and not blocking:
            return False
        if not locked:
            # Lock acquire doesn't take a timeout, so simulate it...
            # XXX: This is not true in Py3 >= 3.2
            try:
                locked = retry(_acquire_lock)
            except RetryFailedError:
                return False
        already_acquired = self.is_acquired
        try:
            gotten = False
            try:
                gotten = retry(
                    self._inner_acquire,
                    blocking=blocking,
                    timeout=timeout,
                    ephemeral=ephemeral,
                )
            except RetryFailedError:
                pass
            except KazooException:
                # if we did ultimately fail, attempt to clean up
                exc_info = sys.exc_info()
                if not already_acquired:
                    self._best_effort_cleanup()
                    self.cancelled = False
                six.reraise(exc_info[0], exc_info[1], exc_info[2])
            if gotten:
                self.is_acquired = gotten
            if not gotten and not already_acquired:
                self._best_effort_cleanup()
            return gotten
        finally:
            self._lock.release()

    def _watch_session(self, state):
        self.wake_event.set()
        return True

    def _inner_acquire(self, blocking, timeout, ephemeral=True):

        # wait until it's our chance to get it..
        if self.is_acquired:
            if not blocking:
                return False
            raise ForceRetryError()

        # make sure our election parent node exists
        if not self.assured_path:
            self._ensure_path()

        node = None
        if self.create_tried:
            node = self._find_node()
        else:
            self.create_tried = True

        if not node:
            node = self.client.create(
                self.create_path, self.data, ephemeral=ephemeral, sequence=True
            )
            # strip off path to node
            node = node[len(self.path) + 1:]

        self.node = node

        while True:
            self.wake_event.clear()

            # bail out with an exception if cancellation has been requested
            if self.cancelled:
                raise CancelledError()

            predecessor = self._get_predecessor(node)
            if predecessor is None:
                return True

            if not blocking:
                return False

            # otherwise we are in the mix. watch predecessor and bide our time
            predecessor = self.path + "/" + predecessor
            self.client.add_listener(self._watch_session)
            try:
                self.client.get(predecessor, self._watch_predecessor)
            except NoNodeError:
                pass  # predecessor has already been deleted
            else:
                self.wake_event.wait(timeout)
                if not self.wake_event.is_set():
                    raise LockTimeout(
                        "Failed to acquire lock on %s after %s seconds"
                        % (self.path, timeout)
                    )
            finally:
                self.client.remove_listener(self._watch_session)

    def _watch_predecessor(self, event):
        self.wake_event.set()

    def _get_predecessor(self, node):
        """returns `node`'s predecessor or None

        Note: This handle the case where the current lock is not a contender
        (e.g. rlock), this and also edge cases where the lock's ephemeral node
        is gone.
        """
        node_sequence = node[len(self.prefix):]
        children = self.client.get_children(self.path)
        found_self = False
        # Filter out the contenders using the computed regex
        contender_matches = []
        for child in children:
            match = self._contenders_re.search(child)
            if match is not None:
                contender_sequence = match.group(1)
                # Only consider contenders with a smaller sequence number.
                # A contender with a smaller sequence number has a higher
                # priority.
                if contender_sequence < node_sequence:
                    contender_matches.append(match)
            if child == node:
                # Remember the node's match object so we can short circuit
                # below.
                found_self = match

        if found_self is False:  # pragma: nocover
            # somehow we aren't in the childrens -- probably we are
            # recovering from a session failure and our ephemeral
            # node was removed.
            raise ForceRetryError()

        if not contender_matches:
            return None

        # Sort the contenders using the sequence number extracted by the regex
        # and return the original string of the predecessor.
        sorted_matches = sorted(contender_matches, key=lambda m: m.groups())
        return sorted_matches[-1].string

    def _find_node(self):
        children = self.client.get_children(self.path)
        for child in children:
            if child.startswith(self.prefix):
                return child
        return None

    def _delete_node(self, node):
        self.client.delete(self.path + "/" + node)

    def _best_effort_cleanup(self):
        try:
            node = self.node or self._find_node()
            if node:
                self._delete_node(node)
        except KazooException:  # pragma: nocover
            pass

    def release(self):
        """Release the lock immediately."""
        return self.client.retry(self._inner_release)

    def _inner_release(self):
        if not self.is_acquired:
            return False

        try:
            self._delete_node(self.node)
        except NoNodeError:  # pragma: nocover
            pass

        self.is_acquired = False
        self.node = None
        return True

    def contenders(self):
        """Return an ordered list of the current contenders for the
        lock.

        .. note::

            If the contenders did not set an identifier, it will appear
            as a blank string.

        """
        # make sure our election parent node exists
        if not self.assured_path:
            self._ensure_path()

        children = self.client.get_children(self.path)
        # We want all contenders, including self (this is especially important
        # for r/w locks). This is similar to the logic of `_get_predecessor`
        # except we include our own pattern.
        all_contenders_re = re.compile(
            r"(?:{patterns})(-?\d{{10}})$".format(
                patterns="|".join(self._exclude_names | {self._NODE_NAME})
            )
        )
        # Filter out the contenders using the computed regex
        contender_matches = []
        for child in children:
            match = all_contenders_re.search(child)
            if match is not None:
                contender_matches.append(match)
        # Sort the contenders using the sequence number extracted by the regex,
        # then extract the original string.
        contender_nodes = [
            match.string
            for match in sorted(contender_matches, key=lambda m: m.groups())
        ]
        # Retrieve all the contender nodes data (preserving order).
        contenders = []
        for node in contender_nodes:
            try:
                data, stat = self.client.get(self.path + "/" + node)
                if data is not None:
                    contenders.append(data.decode("utf-8"))
            except NoNodeError:  # pragma: nocover
                pass

        return contenders

    def __enter__(self):
        self.acquire()

    def __exit__(self, exc_type, exc_value, traceback):
        self.release() class ReadLock(Lock):
    """Kazoo Read Lock

    Example usage with a :class:`~kazoo.client.KazooClient` instance:

    .. code-block:: python

        zk = KazooClient()
        zk.start()
        lock = zk.ReadLock("/lockpath", "my-identifier")
        with lock:  # blocks waiting for outstanding writers
            # do something with the lock

    The lock path passed to WriteLock and ReadLock must match for them to
    communicate.  The read lock blocks if it is held by any writers,
    but multiple readers may hold the lock.

    Note: This lock is not *re-entrant*. Repeated calls after already
    acquired will block.

    This is the read-side of a shared lock.  See :class:`Lock` for a
    standard exclusive lock and :class:`WriteLock` for the write-side of a
    shared lock.

    """

    _NODE_NAME = "__rlock__"
    _EXCLUDE_NAMES = ["__lock__"] class WriteLock(Lock):
    """Kazoo Write Lock

    Example usage with a :class:`~kazoo.client.KazooClient` instance:

    .. code-block:: python

        zk = KazooClient()
        zk.start()
        lock = zk.WriteLock("/lockpath", "my-identifier")
        with lock:  # blocks waiting for lock acquisition
            # do something with the lock

    The lock path passed to WriteLock and ReadLock must match for them to
    communicate.  The write lock can not be acquired if it is held by
    any readers or writers.

    Note: This lock is not *re-entrant*. Repeated calls after already
    acquired will block.

    This is the write-side of a shared lock.  See :class:`Lock` for a
    standard exclusive lock and :class:`ReadLock` for the read-side of a
    shared lock.

    """

    _NODE_NAME = "__lock__"
    _EXCLUDE_NAMES = ["__lock__", "__rlock__"] class Semaphore(object):
    """A Zookeeper-based Semaphore

    This synchronization primitive operates in the same manner as the
    Python threading version only uses the concept of leases to
    indicate how many available leases are available for the lock
    rather than counting.

    Note: This lock is not meant to be *re-entrant*.

    Example:

    .. code-block:: python

        zk = KazooClient()
        semaphore = zk.Semaphore("/leasepath", "my-identifier")
        with semaphore:  # blocks waiting for lock acquisition
            # do something with the semaphore

    .. warning::

        This class stores the allowed max_leases as the data on the
        top-level semaphore node. The stored value is checked once
        against the max_leases of each instance. This check is
        performed when acquire is called the first time. The semaphore
        node needs to be deleted to change the allowed leases.

    .. versionadded:: 0.6
        The Semaphore class.

    .. versionadded:: 1.1
        The max_leases check.

    """

    def __init__(self, client, path, identifier=None, max_leases=1):
        """Create a Kazoo Lock

        :param client: A :class:`~kazoo.client.KazooClient` instance.
        :param path: The semaphore path to use.
        :param identifier: Name to use for this lock contender. This
                           can be useful for querying to see who the
                           current lock contenders are.
        :param max_leases: The maximum amount of leases available for
                           the semaphore.

        """
        # Implementation notes about how excessive thundering herd
        # and watches are avoided
        # - A node (lease pool) holds children for each lease in use
        # - A lock is acquired for a process attempting to acquire a
        #   lease. If a lease is available, the ephemeral node is
        #   created in the lease pool and the lock is released.
        # - Only the lock holder watches for children changes in the
        #   lease pool
        self.client = client
        self.path = path

        # some data is written to the node. this can be queried via
        # contenders() to see who is contending for the lock
        self.data = str(identifier or "").encode("utf-8")
        self.max_leases = max_leases
        self.wake_event = client.handler.event_object()

        self.create_path = self.path + "/" + uuid.uuid4().hex
        self.lock_path = path + "-" + "__lock__"
        self.is_acquired = False
        self.assured_path = False
        self.cancelled = False
        self._session_expired = False

    def _ensure_path(self):
        result = self.client.ensure_path(self.path)
        self.assured_path = True
        if result is True:
            # node did already exist
            data, _ = self.client.get(self.path)
            try:
                leases = int(data.decode("utf-8"))
            except (ValueError, TypeError):
                # ignore non-numeric data, maybe the node data is used
                # for other purposes
                pass
            else:
                if leases != self.max_leases:
                    raise ValueError(
                        "Inconsistent max leases: %s, expected: %s"
                        % (leases, self.max_leases)
                    )
        else:
            self.client.set(self.path, str(self.max_leases).encode("utf-8"))

    def cancel(self):
        """Cancel a pending semaphore acquire."""
        self.cancelled = True
        self.wake_event.set()

    def acquire(self, blocking=True, timeout=None):
        """Acquire the semaphore. By defaults blocks and waits forever.

        :param blocking: Block until semaphore is obtained or
                         return immediately.
        :type blocking: bool
        :param timeout: Don't wait forever to acquire the semaphore.
        :type timeout: float or None

        :returns: Was the semaphore acquired?
        :rtype: bool

        :raises:
            ValueError if the max_leases value doesn't match the
            stored value.

            :exc:`~kazoo.exceptions.LockTimeout` if the semaphore
            wasn't acquired within `timeout` seconds.

        .. versionadded:: 1.1
            The blocking, timeout arguments and the max_leases check.
        """
        # If the semaphore had previously been canceled, make sure to
        # reset that state.
        self.cancelled = False

        try:
            self.is_acquired = self.client.retry(
                self._inner_acquire, blocking=blocking, timeout=timeout
            )
        except KazooException:
            # if we did ultimately fail, attempt to clean up
            self._best_effort_cleanup()
            self.cancelled = False
            raise

        return self.is_acquired

    def _inner_acquire(self, blocking, timeout=None):
        """Inner loop that runs from the top anytime a command hits a
        retryable Zookeeper exception."""
        self._session_expired = False
        self.client.add_listener(self._watch_session)

        if not self.assured_path:
            self._ensure_path()

        # Do we already have a lease?
        if self.client.exists(self.create_path):
            return True

        w = _Watch(duration=timeout)
        w.start()
        lock = self.client.Lock(self.lock_path, self.data)
        try:
            gotten = lock.acquire(blocking=blocking, timeout=w.leftover())
            if not gotten:
                return False
            while True:
                self.wake_event.clear()

                # Attempt to grab our lease...
                if self._get_lease():
                    return True

                if blocking:
                    # If blocking, wait until self._watch_lease_change() is
                    # called before returning
                    self.wake_event.wait(w.leftover())
                    if not self.wake_event.is_set():
                        raise LockTimeout(
                            "Failed to acquire semaphore on %s"
                            " after %s seconds" % (self.path, timeout)
                        )
                else:
                    return False
        finally:
            lock.release()

    def _watch_lease_change(self, event):
        self.wake_event.set()

    def _get_lease(self, data=None):
        # Make sure the session is still valid
        if self._session_expired:
            raise ForceRetryError("Retry on session loss at top")

        # Make sure that the request hasn't been canceled
        if self.cancelled:
            raise CancelledError("Semaphore cancelled")

        # Get a list of the current potential lock holders. If they change,
        # notify our wake_event object. This is used to unblock a blocking
        # self._inner_acquire call.
        children = self.client.get_children(
            self.path, self._watch_lease_change
        )

        # If there are leases available, acquire one
        if len(children) < self.max_leases:
            self.client.create(self.create_path, self.data, ephemeral=True)

        # Check if our acquisition was successful or not. Update our state.
        if self.client.exists(self.create_path):
            self.is_acquired = True
        else:
            self.is_acquired = False

        # Return current state
        return self.is_acquired

    def _watch_session(self, state):
        if state == KazooState.LOST:
            self._session_expired = True
            self.wake_event.set()

            # Return true to de-register
            return True

    def _best_effort_cleanup(self):
        try:
            self.client.delete(self.create_path)
        except KazooException:  # pragma: nocover
            pass

    def release(self):
        """Release the lease immediately."""
        return self.client.retry(self._inner_release)

    def _inner_release(self):
        if not self.is_acquired:
            return False
        try:
            self.client.delete(self.create_path)
        except NoNodeError:  # pragma: nocover
            pass
        self.is_acquired = False
        return True

    def lease_holders(self):
        """Return an unordered list of the current lease holders.

        .. note::

            If the lease holder did not set an identifier, it will
            appear as a blank string.

        """
        if not self.client.exists(self.path):
            return []

        children = self.client.get_children(self.path)

        lease_holders = []
        for child in children:
            try:
                data, stat = self.client.get(self.path + "/" + child)
                lease_holders.append(data.decode("utf-8"))
            except NoNodeError:  # pragma: nocover
                pass
        return lease_holders

    def __enter__(self):
        self.acquire()

    def __exit__(self, exc_type, exc_value, traceback):
        self.release() class SetPartitioner(object):
    """Partitions a set amongst members of a party

    This class will partition a set amongst members of a party such
    that each member will be given zero or more items of the set and
    each set item will be given to a single member. When new members
    enter or leave the party, the set will be re-partitioned amongst
    the members.

    When the :class:`SetPartitioner` enters the
    :attr:`~PartitionState.FAILURE` state, it is unrecoverable
    and a new :class:`SetPartitioner` should be created.

    Example:

    .. code-block:: python

        from kazoo.client import KazooClient
        client = KazooClient()
        client.start()

        qp = client.SetPartitioner(
            path='/work_queues', set=('queue-1', 'queue-2', 'queue-3'))

        while 1:
            if qp.failed:
                raise Exception("Lost or unable to acquire partition")
            elif qp.release:
                qp.release_set()
            elif qp.acquired:
                for partition in qp:
                    # Do something with each partition
            elif qp.allocating:
                qp.wait_for_acquire()

    **State Transitions**

    When created, the :class:`SetPartitioner` enters the
    :attr:`PartitionState.ALLOCATING` state.

    :attr:`~PartitionState.ALLOCATING` ->
    :attr:`~PartitionState.ACQUIRED`

        Set was partitioned successfully, the partition list assigned
        is accessible via list/iter methods or calling list() on the
        :class:`SetPartitioner` instance.

    :attr:`~PartitionState.ALLOCATING` ->
    :attr:`~PartitionState.FAILURE`

        Allocating the set failed either due to a Zookeeper session
        expiration, or failure to acquire the items of the set within
        the timeout period.

    :attr:`~PartitionState.ACQUIRED` ->
    :attr:`~PartitionState.RELEASE`

        The members of the party have changed, and the set needs to be
        repartitioned. :meth:`SetPartitioner.release` should be called
        as soon as possible.

    :attr:`~PartitionState.ACQUIRED` ->
    :attr:`~PartitionState.FAILURE`

        The current partition was lost due to a Zookeeper session
        expiration.

    :attr:`~PartitionState.RELEASE` ->
    :attr:`~PartitionState.ALLOCATING`

        The current partition was released and is being re-allocated.

    """
    def __init__(self, client, path, set, partition_func=None,
                 identifier=None, time_boundary=30, max_reaction_time=1,
                 state_change_event=None):
        """Create a :class:`~SetPartitioner` instance

        :param client: A :class:`~kazoo.client.KazooClient` instance.
        :param path: The partition path to use.
        :param set: The set of items to partition.
        :param partition_func: A function to use to decide how to
                               partition the set.
        :param identifier: An identifier to use for this member of the
                           party when participating. Defaults to the
                           hostname + process id.
        :param time_boundary: How long the party members must be stable
                              before allocation can complete.
        :param max_reaction_time: Maximum reaction time for party members
                                  change.
        :param state_change_event: An optional Event object that will be set
                                   on every state change.

        """
        # Used to differentiate two states with the same names in time
        self.state_id = 0
        self.state = PartitionState.ALLOCATING
        self.state_change_event = state_change_event or \
            client.handler.event_object()

        self._client = client
        self._path = path
        self._set = set
        self._partition_set = []
        self._partition_func = partition_func or self._partitioner
        self._identifier = identifier or '%s-%s' % (
            socket.getfqdn(), os.getpid())
        self._locks = []
        self._lock_path = '/'.join([path, 'locks'])
        self._party_path = '/'.join([path, 'party'])
        self._time_boundary = time_boundary
        self._max_reaction_time = max_reaction_time

        self._acquire_event = client.handler.event_object()

        # Create basic path nodes
        client.ensure_path(path)
        client.ensure_path(self._lock_path)
        client.ensure_path(self._party_path)

        # Join the party
        self._party = client.ShallowParty(self._party_path,
                                          identifier=self._identifier)
        self._party.join()

        self._state_change = client.handler.rlock_object()
        client.add_listener(self._establish_sessionwatch)

        # Now watch the party and set the callback on the async result
        # so we know when we're ready
        self._child_watching(self._allocate_transition, client_handler=True)

    def __iter__(self):
        """Return the partitions in this partition set"""
        for partition in self._partition_set:
            yield partition

    @property
    def failed(self):
        """Corresponds to the :attr:`PartitionState.FAILURE` state"""
        return self.state == PartitionState.FAILURE

    @property
    def release(self):
        """Corresponds to the :attr:`PartitionState.RELEASE` state"""
        return self.state == PartitionState.RELEASE

    @property
    def allocating(self):
        """Corresponds to the :attr:`PartitionState.ALLOCATING`
        state"""
        return self.state == PartitionState.ALLOCATING

    @property
    def acquired(self):
        """Corresponds to the :attr:`PartitionState.ACQUIRED` state"""
        return self.state == PartitionState.ACQUIRED

    def wait_for_acquire(self, timeout=30):
        """Wait for the set to be partitioned and acquired

        :param timeout: How long to wait before returning.
        :type timeout: int

        """
        self._acquire_event.wait(timeout)

    def release_set(self):
        """Call to release the set

        This method begins the step of allocating once the set has
        been released.

        """
        self._release_locks()
        if self._locks:  # pragma: nocover
            # This shouldn't happen, it means we couldn't release our
            # locks, abort
            self._fail_out()
            return
        else:
            with self._state_change:
                if self.failed:
                    return
                self._set_state(PartitionState.ALLOCATING)
        self._child_watching(self._allocate_transition, client_handler=True)

    def finish(self):
        """Call to release the set and leave the party"""
        self._release_locks()
        self._fail_out()

    def _fail_out(self):
        with self._state_change:
            self._set_state(PartitionState.FAILURE)
        if self._party.participating:
            try:
                self._party.leave()
            except KazooException:  # pragma: nocover
                pass

    def _allocate_transition(self, result):
        """Called when in allocating mode, and the children settled"""

        # Did we get an exception waiting for children to settle?
        if result.exception:  # pragma: nocover
            self._fail_out()
            return

        children, async_result = result.get()
        children_changed = self._client.handler.event_object()

        def updated(result):
            with self._state_change:
                children_changed.set()
                if self.acquired:
                    self._set_state(PartitionState.RELEASE)

        with self._state_change:
            # We can lose connection during processing the event
            if not self.allocating:
                return

            # Remember the state ID to check later for race conditions
            state_id = self.state_id

            # updated() will be called when children change
            async_result.rawlink(updated)

        # Check whether the state has changed during the lock acquisition
        # and abort the process if so.
        def abort_if_needed():
            if self.state_id == state_id:
                if children_changed.is_set():
                    # The party has changed. Repartitioning...
                    self._abort_lock_acquisition()
                    return True
                else:
                    return False
            else:
                if self.allocating or self.acquired:
                    # The connection was lost and user initiated a new
                    # allocation process. Abort it to eliminate race
                    # conditions with locks.
                    with self._state_change:
                        self._set_state(PartitionState.RELEASE)

                return True

        # Split up the set
        partition_set = self._partition_func(
            self._identifier, list(self._party), self._set)

        # Proceed to acquire locks for the working set as needed
        for member in partition_set:
            lock = self._client.Lock(self._lock_path + '/' + str(member))

            while True:
                try:
                    # We mustn't lock without timeout because in that case we
                    # can get a deadlock if the party state will change during
                    # lock acquisition.
                    lock.acquire(timeout=self._max_reaction_time)
                except LockTimeout:
                    if abort_if_needed():
                        return
                except KazooException:
                    return self.finish()
                else:
                    break

            self._locks.append(lock)

            if abort_if_needed():
                return

        # All locks acquired. Time for state transition.
        with self._state_change:
            if self.state_id == state_id and not children_changed.is_set():
                self._partition_set = partition_set
                self._set_state(PartitionState.ACQUIRED)
                self._acquire_event.set()
                return

        if not abort_if_needed():
            # This mustn't happen. Means a logical error.
            self._fail_out()

    def _release_locks(self):
        """Attempt to completely remove all the locks"""
        self._acquire_event.clear()
        for lock in self._locks[:]:
            try:
                lock.release()
            except KazooException:  # pragma: nocover
                # We proceed to remove as many as possible, and leave
                # the ones we couldn't remove
                pass
            else:
                self._locks.remove(lock)

    def _abort_lock_acquisition(self):
        """Called during lock acquisition if a party change occurs"""

        self._release_locks()

        if self._locks:
            # This shouldn't happen, it means we couldn't release our
            # locks, abort
            self._fail_out()
            return

        self._child_watching(self._allocate_transition, client_handler=True)

    def _child_watching(self, func=None, client_handler=False):
        """Called when children are being watched to stabilize

        This actually returns immediately, child watcher spins up a
        new thread/greenlet and waits for it to stabilize before
        any callbacks might run.

        :param client_handler: If True, deliver the result using the
                               client's event handler.
        """
        watcher = PatientChildrenWatch(self._client, self._party_path,
                                       self._time_boundary)
        asy = watcher.start()
        if func is not None:
            # We spin up the function in a separate thread/greenlet
            # to ensure that the rawlink's it might use won't be
            # blocked
            if client_handler:
                func = partial(self._client.handler.spawn, func)
            asy.rawlink(func)
        return asy

    def _establish_sessionwatch(self, state):
        """Register ourself to listen for session events, we shut down
        if we become lost"""
        with self._state_change:
            if self.failed:
                pass
            elif state == KazooState.LOST:
                self._client.handler.spawn(self._fail_out)
            elif not self.release:
                self._set_state(PartitionState.RELEASE)

        return state == KazooState.LOST

    def _partitioner(self, identifier, members, partitions):
        # Ensure consistent order of partitions/members
        all_partitions = sorted(partitions)
        workers = sorted(members)

        i = workers.index(identifier)
        # Now return the partition list starting at our location and
        # skipping the other workers
        return all_partitions[i::len(workers)]

    def _set_state(self, state):
        self.state = state
        self.state_id += 1
        self.state_change_event.set() class Queue(BaseQueue):
    """A distributed queue with optional priority support.

    This queue does not offer reliable consumption. An entry is removed
    from the queue prior to being processed. So if an error occurs, the
    consumer has to re-queue the item or it will be lost.

    """

    prefix = "entry-"

    def __init__(self, client, path):
        """
        :param client: A :class:`~kazoo.client.KazooClient` instance.
        :param path: The queue path to use in ZooKeeper.
        """
        super(Queue, self).__init__(client, path)
        self._children = []

    def __len__(self):
        """Return queue size."""
        return super(Queue, self).__len__()

    def get(self):
        """
        Get item data and remove an item from the queue.

        :returns: Item data or None.
        :rtype: bytes
        """
        self._ensure_paths()
        return self.client.retry(self._inner_get)

    def _inner_get(self):
        if not self._children:
            self._children = self.client.retry(
                self.client.get_children, self.path)
            self._children = sorted(self._children)
        if not self._children:
            return None
        name = self._children[0]
        try:
            data, stat = self.client.get(self.path + "/" + name)
            self.client.delete(self.path + "/" + name)
        except NoNodeError:  # pragma: nocover
            # the first node has vanished in the meantime, try to
            # get another one
            self._children = []
            raise ForceRetryError()

        self._children.pop(0)
        return data

    def put(self, value, priority=100):
        """Put an item into the queue.

        :param value: Byte string to put into the queue.
        :param priority:
            An optional priority as an integer with at most 3 digits.
            Lower values signify higher priority.
        """
        self._check_put_arguments(value, priority)
        self._ensure_paths()
        path = '{path}/{prefix}{priority:03d}-'.format(
            path=self.path, prefix=self.prefix, priority=priority)
        self.client.create(path, value, sequence=True) class LockingQueue(BaseQueue):
    """A distributed queue with priority and locking support.

    Upon retrieving an entry from the queue, the entry gets locked with an
    ephemeral node (instead of deleted). If an error occurs, this lock gets
    released so that others could retake the entry. This adds a little penalty
    as compared to :class:`Queue` implementation.

    The user should call the :meth:`LockingQueue.get` method first to lock and
    retrieve the next entry. When finished processing the entry, a user should
    call the :meth:`LockingQueue.consume` method that will remove the entry
    from the queue.

    This queue will not track connection status with ZooKeeper. If a node locks
    an element, then loses connection with ZooKeeper and later reconnects, the
    lock will probably be removed by Zookeeper in the meantime, but a node
    would still think that it holds a lock. The user should check the
    connection status with Zookeeper or call :meth:`LockingQueue.holds_lock`
    method that will check if a node still holds the lock.

    .. note::
        :class:`LockingQueue` requires ZooKeeper 3.4 or above, since it is
        using transactions.
    """
    lock = "/taken"
    entries = "/entries"
    entry = "entry"

    def __init__(self, client, path):
        """
        :param client: A :class:`~kazoo.client.KazooClient` instance.
        :param path: The queue path to use in ZooKeeper.
        """
        super(LockingQueue, self).__init__(client, path)
        self.id = uuid.uuid4().hex.encode()
        self.processing_element = None
        self._lock_path = self.path + self.lock
        self._entries_path = self.path + self.entries
        self.structure_paths = (self._lock_path, self._entries_path)

    def __len__(self):
        """Returns the current length of the queue.

        :returns: queue size (includes locked entries count).
        """
        return super(LockingQueue, self).__len__()

    def put(self, value, priority=100):
        """Put an entry into the queue.

        :param value: Byte string to put into the queue.
        :param priority:
            An optional priority as an integer with at most 3 digits.
            Lower values signify higher priority.

        """
        self._check_put_arguments(value, priority)
        self._ensure_paths()

        self.client.create(
            "{path}/{prefix}-{priority:03d}-".format(
                path=self._entries_path,
                prefix=self.entry,
                priority=priority),
            value, sequence=True)

    def put_all(self, values, priority=100):
        """Put several entries into the queue. The action only succeeds
        if all entries where put into the queue.

        :param values: A list of values to put into the queue.
        :param priority:
            An optional priority as an integer with at most 3 digits.
            Lower values signify higher priority.

        """
        if not isinstance(values, list):
            raise TypeError("values must be a list of byte strings")
        if not isinstance(priority, int):
            raise TypeError("priority must be an int")
        elif priority < 0 or priority > 999:
            raise ValueError("priority must be between 0 and 999")
        self._ensure_paths()

        with self.client.transaction() as transaction:
            for value in values:
                if not isinstance(value, bytes):
                    raise TypeError("value must be a byte string")
                transaction.create(
                    "{path}/{prefix}-{priority:03d}-".format(
                        path=self._entries_path,
                        prefix=self.entry,
                        priority=priority),
                    value, sequence=True)

    def get(self, timeout=None):
        """Locks and gets an entry from the queue. If a previously got entry
        was not consumed, this method will return that entry.

        :param timeout:
            Maximum waiting time in seconds. If None then it will wait
            until an entry appears in the queue.
        :returns: A locked entry value or None if the timeout was reached.
        :rtype: bytes
        """
        self._ensure_paths()
        if self.processing_element is not None:
            return self.processing_element[1]
        else:
            return self._inner_get(timeout)

    def holds_lock(self):
        """Checks if a node still holds the lock.

        :returns: True if a node still holds the lock, False otherwise.
        :rtype: bool
        """
        if self.processing_element is None:
            return False
        lock_id, _ = self.processing_element
        lock_path = "{path}/{id}".format(path=self._lock_path, id=lock_id)
        self.client.sync(lock_path)
        value, stat = self.client.retry(self.client.get, lock_path)
        return value == self.id

    def consume(self):
        """Removes a currently processing entry from the queue.

        :returns: True if element was removed successfully, False otherwise.
        :rtype: bool
        """
        if self.processing_element is not None and self.holds_lock():
            id_, value = self.processing_element
            with self.client.transaction() as transaction:
                transaction.delete("{path}/{id}".format(
                    path=self._entries_path,
                    id=id_))
                transaction.delete("{path}/{id}".format(
                    path=self._lock_path,
                    id=id_))
            self.processing_element = None
            return True
        else:
            return False

    def release(self):
        """Removes the lock from currently processed item without consuming it.

        :returns: True if the lock was removed successfully, False otherwise.
        :rtype: bool

        """
        if self.processing_element is not None and self.holds_lock():
            id_, value = self.processing_element
            with self.client.transaction() as transaction:
                transaction.delete("{path}/{id}".format(
                    path=self._lock_path,
                    id=id_))
            self.processing_element = None
            return True
        else:
            return False

    def _inner_get(self, timeout):
        flag = self.client.handler.event_object()
        lock = self.client.handler.lock_object()
        canceled = False
        value = []

        def check_for_updates(event):
            if event is not None and event.type != EventType.CHILD:
                return
            with lock:
                if canceled or flag.is_set():
                    return
                values = self.client.retry(
                    self.client.get_children,
                    self._entries_path,
                    check_for_updates)
                taken = self.client.retry(
                    self.client.get_children,
                    self._lock_path,
                    check_for_updates)
                available = self._filter_locked(values, taken)
                if len(available) > 0:
                    ret = self._take(available[0])
                    if ret is not None:
                        # By this time, no one took the task
                        value.append(ret)
                        flag.set()

        check_for_updates(None)
        retVal = None
        flag.wait(timeout)
        with lock:
            canceled = True
            if len(value) > 0:
                # We successfully locked an entry
                self.processing_element = value[0]
                retVal = value[0][1]
        return retVal

    def _filter_locked(self, values, taken):
        taken = set(taken)
        available = sorted(values)
        return (available if len(taken) == 0 else
                [x for x in available if x not in taken])

    def _take(self, id_):
        try:
            self.client.create(
                "{path}/{id}".format(
                    path=self._lock_path,
                    id=id_),
                self.id,
                ephemeral=True)
        except NodeExistsError:
            # Item is already locked
            return None

        try:
            value, stat = self.client.retry(
                self.client.get,
                "{path}/{id}".format(path=self._entries_path, id=id_))
        except NoNodeError:
            # Item is already consumed
            self.client.delete(
                "{path}/{id}".format(
                    path=self._lock_path,
                    id=id_))
            return None
        return (id_, value) class ChildrenWatch(object):
    """Watches a node for children updates and calls the specified
    function each time it changes

    The function will also be called the very first time its
    registered to get children.

    Returning `False` from the registered function will disable future
    children change calls. If the client connection is closed (using
    the close command), the ChildrenWatch will no longer get updates.

    if send_event=True in __init__, then the function will always be
    called with second parameter, ``event``. Upon initial call or when
    recovering a lost session the ``event`` is always ``None``.
    Otherwise it's a :class:`~kazoo.prototype.state.WatchedEvent`
    instance.

    Example with client:

    .. code-block:: python

        @client.ChildrenWatch('/path/to/watch')
        def my_func(children):
            print "Children are %s" % children

        # Above function is called immediately and prints children

    """
    def __init__(self, client, path, func=None,
                 allow_session_lost=True, send_event=False):
        """Create a children watcher for a path

        :param client: A zookeeper client.
        :type client: :class:`~kazoo.client.KazooClient`
        :param path: The path to watch for children on.
        :type path: str
        :param func: Function to call initially and every time the
                     children change. `func` will be called with a
                     single argument, the list of children.
        :type func: callable
        :param allow_session_lost: Whether the watch should be
                                   re-registered if the zookeeper
                                   session is lost.
        :type allow_session_lost: bool
        :type send_event: bool
        :param send_event: Whether the function should be passed the
                           event sent by ZooKeeper or None upon
                           initialization (see class documentation)

        The path must already exist for the children watcher to
        run.

        """
        self._client = client
        self._path = path
        self._func = func
        self._send_event = send_event
        self._stopped = False
        self._watch_established = False
        self._allow_session_lost = allow_session_lost
        self._run_lock = client.handler.lock_object()
        self._prior_children = None
        self._used = False

        # Register our session listener if we're going to resume
        # across session losses
        if func is not None:
            self._used = True
            if allow_session_lost:
                self._client.add_listener(self._session_watcher)
            self._get_children()

    def __call__(self, func):
        """Callable version for use as a decorator

        :param func: Function to call initially and every time the
                     children change. `func` will be called with a
                     single argument, the list of children.
        :type func: callable

        """
        if self._used:
            raise KazooException(
                "A function has already been associated with this "
                "ChildrenWatch instance.")

        self._func = func

        self._used = True
        if self._allow_session_lost:
            self._client.add_listener(self._session_watcher)
        self._get_children()
        return func

    @_ignore_closed
    def _get_children(self, event=None):
        with self._run_lock:  # Ensure this runs one at a time
            if self._stopped:
                return

            try:
                children = self._client.retry(self._client.get_children,
                                              self._path, self._watcher)
            except NoNodeError:
                self._stopped = True
                return

            if not self._watch_established:
                self._watch_established = True

                if self._prior_children is not None and \
                   self._prior_children == children:
                    return

            self._prior_children = children

            try:
                if self._send_event:
                    result = self._func(children, event)
                else:
                    result = self._func(children)
                if result is False:
                    self._stopped = True
                    self._func = None
                    if self._allow_session_lost:
                        self._client.remove_listener(self._session_watcher)
            except Exception as exc:
                log.exception(exc)
                raise

    def _watcher(self, event):
        if event.type != "NONE":
            self._get_children(event)

    def _session_watcher(self, state):
        if state in (KazooState.LOST, KazooState.SUSPENDED):
            self._watch_established = False
        elif (state == KazooState.CONNECTED and
              not self._watch_established and not self._stopped):
            self._client.handler.spawn(self._get_children) class DataWatch(object):
    """Watches a node for data updates and calls the specified
    function each time it changes

    The function will also be called the very first time its
    registered to get the data.

    Returning `False` from the registered function will disable future
    data change calls. If the client connection is closed (using the
    close command), the DataWatch will no longer get updates.

    If the function supplied takes three arguments, then the third one
    will be a :class:`~kazoo.protocol.states.WatchedEvent`. It will
    only be set if the change to the data occurs as a result of the
    server notifying the watch that there has been a change. Events
    like reconnection or the first call will not include an event.

    If the node does not exist, then the function will be called with
    ``None`` for all values.

    .. tip::

        Because :class:`DataWatch` can watch nodes that don't exist, it
        can be used alternatively as a higher-level Exists watcher that
        survives reconnections and session loss.

    Example with client:

    .. code-block:: python

        @client.DataWatch('/path/to/watch')
        def my_func(data, stat):
            print("Data is %s" % data)
            print("Version is %s" % stat.version)

        # Above function is called immediately and prints

        # Or if you want the event object
        @client.DataWatch('/path/to/watch')
        def my_func(data, stat, event):
            print("Data is %s" % data)
            print("Version is %s" % stat.version)
            print("Event is %s" % event)

    .. versionchanged:: 1.2

        DataWatch now ignores additional arguments that were previously
        passed to it and warns that they are no longer respected.

    """
    def __init__(self, client, path, func=None, *args, **kwargs):
        """Create a data watcher for a path

        :param client: A zookeeper client.
        :type client: :class:`~kazoo.client.KazooClient`
        :param path: The path to watch for data changes on.
        :type path: str
        :param func: Function to call initially and every time the
                     node changes. `func` will be called with a
                     tuple, the value of the node and a
                     :class:`~kazoo.client.ZnodeStat` instance.
        :type func: callable

        """
        self._client = client
        self._path = path
        self._func = func
        self._stopped = False
        self._run_lock = client.handler.lock_object()
        self._version = None
        self._retry = KazooRetry(max_tries=None,
                                 sleep_func=client.handler.sleep_func)
        self._include_event = None
        self._ever_called = False
        self._used = False

        if args or kwargs:
            warnings.warn('Passing additional arguments to DataWatch is'
                          ' deprecated. ignore_missing_node is now assumed '
                          ' to be True by default, and the event will be '
                          ' sent if the function can handle receiving it',
                          DeprecationWarning, stacklevel=2)

        # Register our session listener if we're going to resume
        # across session losses
        if func is not None:
            self._used = True
            self._client.add_listener(self._session_watcher)
            self._get_data()

    def __call__(self, func):
        """Callable version for use as a decorator

        :param func: Function to call initially and every time the
                     data changes. `func` will be called with a
                     tuple, the value of the node and a
                     :class:`~kazoo.client.ZnodeStat` instance.
        :type func: callable

        """
        if self._used:
            raise KazooException(
                "A function has already been associated with this "
                "DataWatch instance.")

        self._func = func

        self._used = True
        self._client.add_listener(self._session_watcher)
        self._get_data()
        return func

    def _log_func_exception(self, data, stat, event=None):
        try:
            # For backwards compatibility, don't send event to the
            # callback unless the send_event is set in constructor
            if not self._ever_called:
                self._ever_called = True
            try:
                result = self._func(data, stat, event)
            except TypeError:
                result = self._func(data, stat)
            if result is False:
                self._stopped = True
                self._func = None
                self._client.remove_listener(self._session_watcher)
        except Exception as exc:
            log.exception(exc)
            raise

    @_ignore_closed
    def _get_data(self, event=None):
        # Ensure this runs one at a time, possible because the session
        # watcher may trigger a run
        with self._run_lock:
            if self._stopped:
                return

            initial_version = self._version

            try:
                data, stat = self._retry(self._client.get,
                                         self._path, self._watcher)
            except NoNodeError:
                data = None

                # This will set 'stat' to None if the node does not yet
                # exist.
                stat = self._retry(self._client.exists, self._path,
                                   self._watcher)
                if stat:
                    self._client.handler.spawn(self._get_data)
                    return

            # No node data, clear out version
            if stat is None:
                self._version = None
            else:
                self._version = stat.mzxid

            # Call our function if its the first time ever, or if the
            # version has changed
            if initial_version != self._version or not self._ever_called:
                self._log_func_exception(data, stat, event)

    def _watcher(self, event):
        self._get_data(event=event)

    def _set_watch(self, state):
        with self._run_lock:
            self._watch_established = state

    def _session_watcher(self, state):
        if state == KazooState.CONNECTED:
            self._client.handler.spawn(self._get_data)
from collections import defaultdict, deque
from functools import partial
from os.path import split
from kazoo.exceptions import (
    AuthFailedError,
    ConfigurationError,
    ConnectionClosedError,
    ConnectionLoss,
    KazooException,
    NoNodeError,
    NodeExistsError,
    SessionExpiredError,
    WriterNotClosedException,
)
from kazoo.handlers.threading import SequentialThreadingHandler
from kazoo.handlers.utils import capture_exceptions, wrap
from kazoo.hosts import collect_hosts
from kazoo.loggingsupport import BLATHER
from kazoo.protocol.connection import ConnectionHandler
from kazoo.protocol.paths import _prefix_root, normpath
from kazoo.protocol.serialization import (
    Auth,
    CheckVersion,
    CloseInstance,
    Create,
    Create2,
    Delete,
    Exists,
    GetChildren,
    GetChildren2,
    GetACL,
    SetACL,
    GetData,
    Reconfig,
    SetData,
    Sync,
    Transaction
)
from kazoo.protocol.states import (
    Callback,
    EventType,
    KazooState,
    KeeperState,
    WatchedEvent
)
from kazoo.retry import KazooRetry
from kazoo.security import ACL, OPEN_ACL_UNSAFE
from kazoo.recipe.barrier import Barrier, DoubleBarrier
from kazoo.recipe.counter import Counter
from kazoo.recipe.election import Election
from kazoo.recipe.lease import NonBlockingLease, MultiNonBlockingLease
from kazoo.recipe.lock import Lock, ReadLock, WriteLock, Semaphore
from kazoo.recipe.partitioner import SetPartitioner
from kazoo.recipe.party import Party, ShallowParty
from kazoo.recipe.queue import Queue, LockingQueue
from kazoo.recipe.watchers import ChildrenWatch, DataWatch
import inspect
import logging
import re
import warnings
import six

            .. code-block:: python

                sasl_options = {
                    'mechanism': 'GSSAPI',
                    'service': 'myzk',                  # optional
                    'principal': 'client@EXAMPLE.COM'   # optional
                }

        :param read_only: Allow connections to read only servers.
        :param randomize_hosts: By default randomize host selection.
        :param connection_retry:
            A :class:`kazoo.retry.KazooRetry` object to use for
            retrying the connection to Zookeeper. Also can be a dict of
            options which will be used for creating one.
        :param command_retry:
            A :class:`kazoo.retry.KazooRetry` object to use for
            the :meth:`KazooClient.retry` method. Also can be a dict of
            options which will be used for creating one.
        :param logger: A custom logger to use instead of the module
            global `log` instance.
        :param keyfile: SSL keyfile to use for authentication
        :param keyfile_password: SSL keyfile password
        :param certfile: SSL certfile to use for authentication
        :param ca: SSL CA file to use for authentication
        :param use_ssl: argument to control whether SSL is used or not
        :param verify_certs: when using SSL, argument to bypass
            certs verification

        Basic Example:

        .. code-block:: python

            zk = KazooClient()
            zk.start()
            children = zk.get_children('/')
            zk.stop()

        As a convenience all recipe classes are available as attributes
        and get automatically bound to the client. For example::

            zk = KazooClient()
            zk.start()
            lock = zk.Lock('/lock_path')

        .. versionadded:: 0.6
            The read_only option. Requires Zookeeper 3.4+

        .. versionadded:: 0.6
            The retry_max_delay option.

        .. versionadded:: 0.6
            The randomize_hosts option.

        .. versionchanged:: 0.8
            Removed the unused watcher argument (was second argument).

        .. versionadded:: 1.2
            The connection_retry, command_retry and logger options.

        .. versionadded:: 2.7
            The sasl_options option.

        """
        self.logger = logger or log

        # Record the handler strategy used
        self.handler = handler if handler else SequentialThreadingHandler()
        if inspect.isclass(self.handler):
            raise ConfigurationError("Handler must be an instance of a class, "
                                     "not the class: %s" % self.handler)

        self.auth_data = auth_data if auth_data else set([])
        self.default_acl = default_acl
        self.randomize_hosts = randomize_hosts
        self.hosts = None
        self.chroot = None
        self.set_hosts(hosts)

        self.use_ssl = use_ssl
        self.verify_certs = verify_certs
        self.certfile = certfile
        self.keyfile = keyfile
        self.keyfile_password = keyfile_password
        self.ca = ca
        # Curator like simplified state tracking, and listeners for
        # state transitions
        self._state = KeeperState.CLOSED
        self.state = KazooState.LOST
        self.state_listeners = set()
        self._child_watchers = defaultdict(set)
        self._data_watchers = defaultdict(set)
        self._reset()
        self.read_only = read_only

        if client_id:
            self._session_id = client_id[0]
            self._session_passwd = client_id[1]
        else:
            self._reset_session()

        # ZK uses milliseconds
        self._session_timeout = int(timeout * 1000)

        # We use events like twitter's client to track current and
        # desired state (connected, and whether to shutdown)
        self._live = self.handler.event_object()
        self._writer_stopped = self.handler.event_object()
        self._stopped = self.handler.event_object()
        self._stopped.set()
        self._writer_stopped.set()

        self.retry = self._conn_retry = None

        if type(connection_retry) is dict:
            self._conn_retry = KazooRetry(**connection_retry)
        elif type(connection_retry) is KazooRetry:
            self._conn_retry = connection_retry

        if type(command_retry) is dict:

            self.retry = KazooRetry(**command_retry)