separate abstract and generic sink handler

src/AbstractSinkHandler.py

@@ -0,0 +1,32 @@
+import threading
+from loguru import logger
+from DataObject import DataObject
+
+
+POISON_PILL = DataObject(name="PoisonPill", topic="kill", payload=None)
+
+
+class AbstractSinkHandler(threading.Thread):
+    def __init__(self, config, inQueue, experiment):
+        super().__init__()
+
+        self.config = config
+        self.inQueue = inQueue
+        self.experiment = experiment
+
+    def run(self):
+        logger.debug("loop started")
+
+        while True:
+            dataObject = self.inQueue.get()
+            if (dataObject == POISON_PILL):
+                logger.debug("swallowed the poison pill")
+                break
+            self.sinkAction(dataObject)            
+
+    def sinkAction(self, dataObject):
+        raise NotImplementedError()
+
+    def stop(self):
+        self.inQueue.put(POISON_PILL)
+        logger.debug("kill flag set")

src/GenericSinkHandler.py

@@ -1,30 +1,9 @@
-import threading
 from loguru import logger
-from DataObject import DataObject
+from AbstractSinkHandler import AbstractSinkHandler
 
-
+class GenericSinkHandler(AbstractSinkHandler):
-POISON_PILL = DataObject(name="PoisonPill", topic="kill", payload=None)
-
-
-class GenericSinkHandler(threading.Thread):
     def __init__(self, config, inQueue, experiment):
-        super().__init__()
+        super().__init__(config, inQueue, experiment)
     
-        self.config = config
+    def sinkAction(self, dataObject):
-        self.inQueue = inQueue
+        logger.info(f"received {dataObject} for experiment {self.experiment}")
-        self.experiment = experiment
-
-    def run(self):
-        logger.debug("GenericSinkHandler loop started")
-
-        while True:
-            dataObject = self.inQueue.get()
-            if (dataObject == POISON_PILL):
-                logger.debug("GenericSinkHandler swallowed the poison pill")
-                break
-            logger.info(f"GenericSinkHandler received {dataObject} for experiment {self.experiment}")
-            
-
-    def stop(self):
-        self.inQueue.put(POISON_PILL)
-        logger.debug("kill flag set")

src/SlaveHandler.py

@@ -2,16 +2,23 @@ from queue import Queue
 from loguru import logger
 from GenericMqttSubscriber import GenericMqttSubscriber
 from GenericSinkHandler import GenericSinkHandler
+import importlib
 
 
 class SlaveHandler(object):
     def __init__(self, config, name, experiment):
         self.config = config
         self.name = name
+        localConfig = self.config[name]
 
         queue = Queue()
         self.mqttHandler = GenericMqttSubscriber(config, self.name, queue)
-        self.sinkHandler = GenericSinkHandler(config, queue, experiment)
+
+        sinkHandlerName = localConfig["sinkHandler"]
+        sinkHandlerModule = importlib.import_module(sinkHandlerName)
+        logger.info(f"sink handler {sinkHandlerName} loaded, about to instantiate it")
+        sinkHandlerClass = getattr(sinkHandlerModule, sinkHandlerName)
+        self.sinkHandler = sinkHandlerClass(config, queue, experiment)
     
     def start(self):
         logger.info(f"Starting slave handler {self.name}")

src/config/iiotFeederConfig.ini

@@ -11,12 +11,15 @@ slaves=dev05 dev06 rig01
 [dev05]
 topic=rd/set01/rig01/dev05/md
 dataObjectName=value
+sinkHandler=GenericSinkHandler
 
 [dev06]
 topic=rd/set01/rig01/dev06/md
 dataObjectName=value
+sinkHandler=GenericSinkHandler
 
 [rig01]
 topic=rd/set01/rig01/log
 dataObjectName=log
+sinkHandler=GenericSinkHandler