@@ -0,0 +1,3453 @@
+%22%22%22Threading Handler%22%22%22%0Aimport Queue%0Aimport threading%0A%0Afrom zope.interface import implementer%0A%0Afrom kazoo.interfaces import IAsyncResult%0Afrom kazoo.interfaces import IHandler%0A%0A# sentinal object%0A_NONE = object()%0A%0A%0Aclass TimeoutError(Exception):%0A    pass%0A%0A%0A@implementer(IAsyncResult)%0Aclass AsyncResult(object):%0A    %22%22%22A one-time event that stores a value or an exception%22%22%22%0A    def __init__(self):%0A        self.value = None%0A        self._exception = _NONE%0A        self._condition = threading.Condition()%0A%0A    def ready(self):%0A        %22%22%22Return true if and only if it holds a value or an exception%22%22%22%0A        return self._exception is not _NONE%0A%0A    def successful(self):%0A        %22%22%22Return true if and only if it is ready and holds a value%22%22%22%0A        return self._exception is None%0A%0A    @property%0A    def exception(self):%0A        if self._exception is not _NONE:%0A            return self._exception%0A%0A    def set(self, value=None):%0A        %22%22%22Store the value. Wake up the waiters.%0A        %22%22%22%0A        with self._condition:%0A            self.value = value%0A            self._exception = None%0A%0A            self._condition.notify_all()%0A%0A    def set_exception(self, exception):%0A        %22%22%22Store the exception. Wake up the waiters.%0A        %22%22%22%0A        with self._condition:%0A            self._exception = exception%0A%0A            self._condition.notify_all()%0A%0A    def get(self, block=True, timeout=None):%0A        %22%22%22Return the stored value or raise the exception.%0A%0A        If there is no value raises Timeout%0A        %22%22%22%0A        with self._condition:%0A            if self._exception is not _NONE:%0A                if self._exception is None:%0A                    return self.value%0A                raise self._exception%0A            elif block:%0A                self._condition.wait(timeout)%0A                if self._exception is not _NONE:%0A                    if self._exception is None:%0A                        return self.value%0A                    raise self._exception%0A%0A            # if we get to this point we timeout%0A            raise TimeoutError()%0A%0A    def get_nowait(self):%0A        %22%22%22Return the value or raise the exception without blocking.%0A%0A        If nothing is available, raises TimeoutError%0A        %22%22%22%0A        return self.get(block=False)%0A%0A%0A@implementer(IHandler)%0Aclass SequentialThreadingHandler(object):%0A    name = %22sequential_threading_handler%22%0A    timeout_exception = TimeoutError%0A%0A    def __init__(self):%0A        self.callback_queue = Queue.Queue()%0A        self.session_queue = Queue.Queue()%0A        self._running = True%0A%0A        # Spawn our worker threads, we have%0A        # - A callback worker for watch events to be called%0A        # - A session worker for session events to be called%0A        self._create_thread_worker(self.callback_queue)%0A        self._create_thread_worker(self.session_queue)%0A%0A    def _create_thread_worker(self, queue):%0A        def thread_worker():%0A            while self._running:%0A                try:%0A                    func = queue.get(timeout=1)%0A                    func()%0A                except Queue.Empty:%0A                    continue%0A        thread = threading.Thread(target=thread_worker)%0A        thread.start()%0A%0A    def async_result(self):%0A        return AsyncResult(self)%0A%0A    def dispatch_callback(self, callback):%0A        if callback.type == 'session':%0A            self.session_queue.put(lambda: callback.func(*callback.args))%0A        else:%0A            self.callback_queue.put(lambda: callback.func(*callback.args))%0A

Add threading handler