KIT-1033 made worker thread init dynamic

Author: oschulzSam

docs/conf.py

@@ -18,6 +18,9 @@
 #
 import os
 import sys
+
+import setup
+
 sys.path.insert(0, os.path.abspath('..'))
 
 import logstash_async
@@ -56,7 +59,7 @@
 # built documents.
 #
 # The short X.Y version.
-version = logstash_async.__version__
+version = setup.VERSION
 # The full version, including alpha/beta/rc tags.
 release = version
 

logstash_async/__init__.py

@@ -1,10 +0,0 @@
-# -*- coding: utf-8 -*-
-#
-# This software may be modified and distributed under the terms
-# of the MIT license.  See the LICENSE file for details.
-
-__version__ = '2.3.2'
-
-# When using an in-memory only cache, this persists the cache through
-# thread failures, shutdowns, and restarts.
-EVENT_CACHE = {}

logstash_async/handler.py

@@ -9,7 +9,6 @@
 from logstash_async.formatter import LogstashFormatter
 from logstash_async.utils import safe_log_via_print
 from logstash_async.worker import LogProcessingWorker
-import logstash_async
 from .transport import HttpTransport
 
 
@@ -28,8 +27,6 @@ class AsynchronousLogstashHandler(Handler):
                       the database. (Given in seconds. Default is None, and disables this feature)
     """
 
-    _worker_thread = None
-
     # ----------------------------------------------------------------------
     # pylint: disable=too-many-arguments
     def __init__(self, host, port,
@@ -43,6 +40,7 @@ def __init__(self, host, port,
         self._transport = transport
         self._event_ttl = event_ttl
         self._encoding = encoding
+        self._worker_thread = None
         self._setup_transport(**kwargs)
 
     # ----------------------------------------------------------------------
@@ -53,10 +51,9 @@ def emit(self, record):
         self._setup_transport()
         self._start_worker_thread()
 
-        # basically same implementation as in logging.handlers.SocketHandler.emit()
         try:
             data = self._format_record(record)
-            AsynchronousLogstashHandler._worker_thread.enqueue_event(data)
+            self._worker_thread.enqueue_event(data)
         except Exception:
             self.handleError(record)
 
@@ -80,23 +77,18 @@ def _start_worker_thread(self):
         if self._worker_thread_is_running():
             return
 
-        AsynchronousLogstashHandler._worker_thread = LogProcessingWorker(
+        self._worker_thread = LogProcessingWorker(
             host=self._host,
             port=self._port,
             transport=self._transport,
             ssl_enable=self._ssl_enable,
-            cache=logstash_async.EVENT_CACHE,
+            cache={},
             event_ttl=self._event_ttl)
-        AsynchronousLogstashHandler._worker_thread.start()
+        self._worker_thread.start()
 
     # ----------------------------------------------------------------------
-    @staticmethod
-    def _worker_thread_is_running():
-        worker_thread = AsynchronousLogstashHandler._worker_thread
-        if worker_thread is not None and worker_thread.is_alive():
-            return True
-
-        return False
+    def _worker_thread_is_running(self):
+        return self._worker_thread is not None and self._worker_thread.is_alive()
 
     # ----------------------------------------------------------------------
     def _format_record(self, record):
@@ -132,15 +124,15 @@ def shutdown(self):
 
     # ----------------------------------------------------------------------
     def _trigger_worker_shutdown(self):
-        AsynchronousLogstashHandler._worker_thread.shutdown()
+        self._worker_thread.shutdown()
 
     # ----------------------------------------------------------------------
     def _wait_for_worker_thread(self):
-        AsynchronousLogstashHandler._worker_thread.join()
+        self._worker_thread.join()
 
     # ----------------------------------------------------------------------
     def _reset_worker_thread(self):
-        AsynchronousLogstashHandler._worker_thread = None
+        self._worker_thread = None
 
     # ----------------------------------------------------------------------
     def _close_transport(self):