threads and queues

MongoWriter.py

@@ -0,0 +1,36 @@
+'''
+Created on 09.06.2015
+
+@author: wn
+'''
+
+
+import threading
+from logger import Logger
+import pymongo
+
+class MongoWriter(threading.Thread):
+    def __init__(self, queue, dbhost, database, collection):
+        threading.Thread.__init__(self)
+        self.queue = queue
+        self.dbhost = dbhost
+        self.database = database
+        self.collection = collection
+        self.setDaemon(True)
+
+    def run(self):
+        while True:
+            try:
+                msg = self.queue.get()
+                # Logger.debug("MongoWriter receives: %s" % msg)
+                mongoClient = pymongo.MongoClient(self.dbhost)
+                db = mongoClient[self.database]
+                res = db[self.collection].insert_one(msg)
+                Logger.debug("MongoWriter inserts: %s" % res.inserted_id)
+            except pymongo.errors.ServerSelectionTimeoutError, e:
+                Logger.log("Exception %s in MongoWriter, run" % str(e))
+                Logger.log("Msg dropped: %s" % msg)
+            except TypeError, e:
+                Logger.log("Exception %s in MongoWriter, run" % str(e))
+            except Exception, e:
+                Logger.log("Unexcepted exception %s in MongoWriter: %s" % (e.__class__.__name__, str(e)))

MqttReceiver.py

@@ -10,46 +10,54 @@ from logger import Logger
 import paho.mqtt.client as mqtt
 import json
 import datetime
+from twisted.spread.pb import Broker
 
-__queue = None
+_queue = None
 
 def on_message(client, userdata, msg):
+    try:
         j = json.loads(msg.payload)
-    now = datetime.datetime.now()
+        now1 = datetime.datetime.now()
+        now = now1.replace(now1.year, now1.month, now1.day, now1.hour, now1.minute, now1.second, 0)
         midnight = now.replace(now.year, now.month, now.day, 0,0,0,0)
         seconds = (now - midnight).seconds
-    j['metadata']['timestamp'] = datetime.datetime.now()
+        j['metadata']['timestamp'] = now
         j['metadata']['seconds'] = seconds
         j['metadata']['day'] = midnight
 
-    try:
+        Logger.debug("MqttReceiver queues: %s" % j)
-        __queue.put_nowait(j)
+    
+        _queue.put_nowait(j)
     except Queue.Full:
         Logger.log("Message %s dropped" % (j))
-
+    except ValueError, e:
-
+        Logger.log("Exception %s in MqttReceiver, on_message" % (str(e)))
 
 
 class MqttReceiver(threading.Thread):
     singleton = None
     
     @classmethod
-    def create(cls, queue):
+    def create(cls, queue, broker, topics):
+        global _queue
         if cls.singleton is None:
-            cls.singleton = MqttReceiver(queue)
+            cls.singleton = MqttReceiver(queue, broker, topics)
-            __queue = queue
+            _queue = queue
         return cls.singleton
             
-    def __init__(self, queue):
+    def __init__(self, queue, broker, topics):
         threading.Thread.__init__(self)
         self.queue = queue
+        self.broker = broker
+        self.topics = topics
         self.setDaemon(True)
 
     def run(self):
         client = mqtt.Client()
         client.on_message = on_message
-        client.connect("mqttbroker", 1883, 60)
+        client.connect(self.broker, 1883, 60)
-        client.subscribe("IoT/Measurement/#")
+        for topic in self.topics:
-        client.subscribe("IoT/WiFiPowerMeter/Measurement")
+            Logger.log("Subscribing on %s" % str(topic))
+            client.subscribe(topic)
         
         client.loop_forever()

logger.py

@@ -1,19 +1,31 @@
-
+from time import gmtime, strftime
 
 class Logger(object):
-    @staticmethod
-    def log(data):
-        print data
-        if Logger.debugFlag:
-            print data
-    
-    @staticmethod
-    def debugEnable():
-        Logger.debugFlag = True
-
-    @staticmethod
-    def debugDisable():
-        Logger.debugFlag = False
-
     debugFlag = False
 
+    @classmethod
+    def log(cls, data):
+        t = strftime("%d %b %Y %H:%M:%S", gmtime())
+        with open(cls.logfile, 'a') as f:
+            f.write("%s %s\n" % (t, data))
+        if cls.debugFlag:
+            print data
+    
+    @classmethod
+    def debug(cls, data):
+        if cls.debugFlag:
+            cls.log(data)
+    
+    @classmethod
+    def debugEnable(cls):
+        cls.debugFlag = True
+
+    @classmethod
+    def debugDisable(cls):
+        cls.debugFlag = False
+
+    @classmethod
+    def openlog(cls, logfile):
+        cls.logfile = logfile
+        
+

mqtt2mongo.py

@@ -4,33 +4,62 @@ Created on 20.05.2015
 @author: wn
 '''
 
-import paho.mqtt.client as mqtt
+from MqttReceiver import MqttReceiver
-import json
+from MongoWriter import MongoWriter
-import pymongo
+import Queue
-import datetime
+import os
+import sys
+from logger import Logger
+import time
 
-def on_message(client, userdata, msg):
+DEBUG = True
-    j = json.loads(msg.payload)
+BACKGROUND = False
-    now = datetime.datetime.now()
+PID_FILE = "/tmp/mqtt2mongo.pid"
-    midnight = now.replace(now.year, now.month, now.day, 0,0,0,0)
+LOG_FILE = "/tmp/mqtt2mongo.log"
-    seconds = (now - midnight).seconds
+BROKER = "mqttbroker"
-    j['metadata']['timestamp'] = datetime.datetime.now()
+TOPICS = ['IoT/Measurement/#', 'IoT/WiFiPowerMeter/Measurement']
-    j['metadata']['seconds'] = seconds
+MONGO_HOST = "172.16.2.15"
-    j['metadata']['day'] = midnight
+MONGO_DATABASE = "test"
-    print(j)
+MONGO_COLLECTION = "test"
-    mongoClient = pymongo.MongoClient('localhost')
-    db = mongoClient.iot
-    r = db.iot.insert_one(j)
 
 
-if __name__ == '__main__':
+if BACKGROUND:
-    print("Starting client ...")
+    pid = os.fork()
+else:
+    pid = 0
 
-    client = mqtt.Client()
+if pid:
-    client.on_message = on_message
+    pidFile = file(PID_FILE , mode='w')
-    client.connect("mqttbroker", 1883, 60)
+    pidFile.write("%i\n" % pid)
-    client.subscribe("IoT/Measurement/#")
+    pidFile.close()
-    client.subscribe("IoT/WiFiPowerMeter/Measurement")
+    sys.exit(0)
 
-    client.loop_forever()
+Logger.openlog(LOG_FILE)
 
+if DEBUG:
+    Logger.debugEnable()
+
+Logger.log("mqtt2mongo starting")
+
+
+
+try:
+    queue = Queue.Queue()
+    Logger.log("queue created ...")
+    
+    mongoWriter = MongoWriter(queue, 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")
+    
+    while (True):
+        time.sleep(10)
+except Exception, e:
+    Logger.log("mqtt2mongo dying from %s, %s" % (str(e), e.msg))
+finally:
+    Logger.log("mqtt2mongo terminating ...")