refactor: Unify queue factory in various handlers

Then every handlers have queue_impl and queue_empty as their attributes.

2 files changed, 9 insertions, 7 deletions

diff --git a/kazoo/handlers/eventlet.py b/kazoo/handlers/eventlet.py
index 1126298..c87898e 100644
--- a/kazoo/handlers/eventlet.py
+++ b/kazoo/handlers/eventlet.py
@@ -76,11 +76,13 @@ class SequentialEventletHandler(object):
 
     """
     name = "sequential_eventlet_handler"
+    queue_impl = green_queue.LightQueue
+    queue_empty = green_queue.Empty
 
     def __init__(self):
         """Create a :class:`SequentialEventletHandler` instance"""
-        self.callback_queue = green_queue.LightQueue()
-        self.completion_queue = green_queue.LightQueue()
+        self.callback_queue = self.queue_impl()
+        self.completion_queue = self.queue_impl()
         self._workers = []
         self._started = False
 
diff --git a/kazoo/handlers/gevent.py b/kazoo/handlers/gevent.py
index 71a4237..551e6a0 100644
--- a/kazoo/handlers/gevent.py
+++ b/kazoo/handlers/gevent.py
@@ -7,8 +7,6 @@ import gevent
 from gevent import socket
 import gevent.event
 import gevent.queue
-from gevent.queue import Empty
-from gevent.queue import Queue
 import gevent.select
 import gevent.thread
 try:
@@ -50,11 +48,13 @@ class SequentialGeventHandler(object):
 
     """
     name = "sequential_gevent_handler"
+    queue_impl = gevent.queue.Queue
+    queue_empty = gevent.queue.Empty
     sleep_func = staticmethod(gevent.sleep)
 
     def __init__(self):
         """Create a :class:`SequentialGeventHandler` instance"""
-        self.callback_queue = Queue()
+        self.callback_queue = self.queue_impl()
         self._running = False
         self._async = None
         self._state_change = Semaphore()
@@ -72,7 +72,7 @@ class SequentialGeventHandler(object):
                     if func is _STOP:
                         break
                     func()
-                except Empty:
+                except self.queue_empty:
                     continue
                 except Exception as exc:
                     log.warning("Exception in worker greenlet")
@@ -110,7 +110,7 @@ class SequentialGeventHandler(object):
                 worker.join()
 
             # Clear the queues
-            self.callback_queue = Queue()  # pragma: nocover
+            self.callback_queue = self.queue_impl()  # pragma: nocover
 
             python2atexit.unregister(self.stop)