Bring QueueListener implementation up to date with changes in stdlib version.

1 files changed, 9 insertions, 16 deletions

diff --git a/logutils/queue.py b/logutils/queue.py
index fea91d8..b4b2151 100644
--- a/logutils/queue.py
+++ b/logutils/queue.py
@@ -109,15 +109,15 @@ class QueueListener(object):
     """
     _sentinel = None
 
-    def __init__(self, queue, *handlers):
+    def __init__(self, queue, *handlers, **kwargs):
         """
         Initialise an instance with the specified queue and
         handlers.
         """
         self.queue = queue
         self.handlers = handlers
-        self._stop = threading.Event()
         self._thread = None
+        self.respect_handler_level = kwargs.get('respect_handler_level', False)
 
     def dequeue(self, block):
         """
@@ -167,7 +167,12 @@ class QueueListener(object):
         """
         record = self.prepare(record)
         for handler in self.handlers:
-            handler.handle(record)
+            if not self.respect_handler_level:
+                process = True
+            else:
+                process = record.levelno >= handler.level
+            if process:
+                handler.handle(record)
 
     def _monitor(self):
         """
@@ -179,20 +184,9 @@ class QueueListener(object):
         """
         q = self.queue
         has_task_done = hasattr(q, 'task_done')
-        while not self._stop.isSet():
-            try:
-                record = self.dequeue(True)
-                if record is self._sentinel:
-                    break
-                self.handle(record)
-                if has_task_done:
-                    q.task_done()
-            except queue.Empty:
-                pass
-        # There might still be records in the queue.
         while True:
             try:
-                record = self.dequeue(False)
+                record = self.dequeue(True)
                 if record is self._sentinel:
                     break
                 self.handle(record)
@@ -218,7 +212,6 @@ class QueueListener(object):
         Note that if you don't call this before your application exits, there
         may be some records still left on the queue, which won't be processed.
         """
-        self._stop.set()
         self.enqueue_sentinel()
         self._thread.join()
         self._thread = None