merged

MeterBusDecoder.py

@@ -0,0 +1,33 @@
+'''
+Created on 09.06.2015
+
+@author: wn
+'''
+
+
+import threading
+from logger import Logger
+import Queue
+
+
+
+
+class MeterBusDecoder(threading.Thread):
+    def __init__(self, inQueue, outQueue):
+        threading.Thread.__init__(self)
+        self.inQueue = inQueue
+        self.outQueue = outQueue
+        self.setDaemon(True)
+
+    def run(self):
+        while True:
+            msg = None
+            try:
+                msg = self.inQueue.get()
+
+                Logger.log("MeterBusDecoder is doing something with %s" % msg)
+
+                self.outQueue.put_nowait(msg)
+            except Queue.Full:
+                Logger.log("MeterBusDecoder queue overrun, drop result of %s" % str(msg)) 
+

mqtt2mongo.py

@@ -6,6 +6,7 @@ Created on 20.05.2015
 
 from MqttReceiver import MqttReceiver
 from MongoWriter import MongoWriter
+from MeterBusDecoder import MeterBusDecoder
 import Queue
 import os
 import sys
@@ -44,16 +45,22 @@ Logger.log("mqtt2mongo starting")
 
 
 try:
-    queue = Queue.Queue()
-    Logger.log("queue created ...")
+    queue1 = Queue.Queue()
+    queue2 = Queue.Queue()
+    Logger.log("queues created ...")
     
-    mongoWriter = MongoWriter(queue, MONGO_HOST, MONGO_DATABASE, MONGO_COLLECTION)
+    mqttReader = MqttReceiver.create(queue1, BROKER, TOPICS)
+    mqttReader.start()
+    Logger.log("MqttReader started ...")
+    
+    meterBusDecoder = MeterBusDecoder(queue1, queue2)
+    meterBusDecoder.start()
+    Logger.log("MeterBusDecoder started ...")
+
+    mongoWriter = MongoWriter(queue2, MONGO_HOST, MONGO_DATABASE, MONGO_COLLECTION)
     mongoWriter.start()
     Logger.log("MongoWriter started ...")
     
-    mqttReader = MqttReceiver.create(queue, BROKER, TOPICS)
-    mqttReader.start()
-    Logger.log("MqttReader started ...")
     
     Logger.log("mqtt2mongo running")