"""gevent Handler"""
from __future__ import absolute_import

import gevent
import gevent.event
import gevent.queue
from zope.interface import implementer

from kazoo.interfaces import IAsyncResult
from kazoo.interfaces import IHandler


@implementer(IAsyncResult)
class AsyncResult(gevent.event.AsyncResult):
    def __init__(self, handler):
        self._handler = handler
        gevent.event.AsyncResult.__init__(self)

    def set(self, value=None):
        # Proxy the set call to the gevent thread
        self._handler.completion_queue.put(
            lambda: gevent.event.AsyncResult.set(self, value)
        )

    def set_exception(self, exception):
        # Proxy the set_exception call to the gevent thread
        self._handler.completion_queue.put(
            lambda: gevent.event.AsyncResult.set_exception(self, exception)
        )


@implementer(IHandler)
class SequentialGeventHandler(object):
    timeout_exception = gevent.event.Timeout

    def __init__(self):
        self.completion_queue = gevent.queue.Queue()
        self.callback_queue = gevent.queue.Queue()
        self.session_queue = gevent.queue.Queue()
        self._running = True

        # Spawn our worker greenlets, we have
        # - A completion worker for when values come back to be set on
        #   the AsyncResult object
        # - A callback worker for watch events to be called
        # - A session worker for session events to be called
        self._create_greenlet_worker(self.completion_queue)
        self._create_greenlet_worker(self.callback_queue)
        self._create_greenlet_worker(self.session_queue)

    def _create_greenlet_worker(self, queue):
        def greenlet_worker():
            while self._running:
                try:
                    func = queue.get(timeout=1)
                    func()
                except gevent.queue.Empty:
                    continue
        gevent.spawn(greenlet_worker)

    def async_result(self):
        return AsyncResult(self)

    def dispatch_callback(self, callback):
        if callback.type == 'session':
            self.session_queue.put(lambda: callback.func(*callback.args))
        else:
            self.callback_queue.put(lambda: callback.func(*callback.args))

@@ -54,16 +54,40 @@
 import%0A%0A
+import fcntl%0Aimport os%0A%0A
 import g
@@ -116,23 +116,65 @@
 ent%0A
-import gevent.q
+from gevent.queue import Empty%0Afrom gevent.queue import Q
 ueue
@@ -296,16 +296,506 @@
 ndler%0A%0A%0A
+# Simple wrapper to os.pipe() - but sets to non-block%0Adef _pipe():%0A    r, w = os.pipe()%0A    fcntl.fcntl(r, fcntl.F_SETFL, os.O_NONBLOCK)%0A    fcntl.fcntl(w, fcntl.F_SETFL, os.O_NONBLOCK)%0A    return r, w%0A%0A_core_pipe_read, _core_pipe_write = _pipe()%0A%0A%0Adef _core_pipe_read_callback(event, evtype):%0A    try:%0A        os.read(event.fd, 1)%0A    except:%0A        pass%0A%0Agevent.core.event(gevent.core.EV_READ %7C gevent.core.EV_PERSIST,%0A                 _core_pipe_read, _core_pipe_read_callback).add()%0A%0A%0A
 @impleme
@@ -1170,16 +1170,90 @@
      )%0A%0A
+        # Wake gevent wait/gets%0A        os.write(_core_pipe_write, '%5C0')%0A%0A
     def 
@@ -1475,16 +1475,90 @@
      )%0A%0A
+        # Wake gevent wait/gets%0A        os.write(_core_pipe_write, '%5C0')%0A%0A
 %0A@implem
@@ -1710,37 +1710,24 @@
 ion_queue = 
-gevent.queue.
 Queue()%0A    
@@ -1748,37 +1748,24 @@
 ack_queue = 
-gevent.queue.
 Queue()%0A    
@@ -1789,29 +1789,16 @@
 queue = 
-gevent.queue.
 Queue()%0A
@@ -2386,16 +2386,149 @@
 unning:%0A
+                # We timeout after 1 and repeat so that we can gracefully%0A                # shutdown if self_running is set to false%0A
         
@@ -2642,21 +2642,8 @@
 ept 
-gevent.queue.
 Empt
@@ -3000,28 +3000,102 @@
 lback.func(*callback.args))%0A
+%0A        # Wake gevent wait/gets%0A        os.write(_core_pipe_write, '%5C0')%0A

Use pipe trick to ensure gevent queue's wake up when set from a separate thread.