persistent queue is persistent now

This commit is contained in:
Wolfgang Hottgenroth 2017-11-14 13:06:05 +01:00
parent c445edf5c2
commit f039c58bdb
3 changed files with 40 additions and 13 deletions

View File

@ -1,7 +1,7 @@
import threading
from logger import Logger
import MySQLdb
import signal
class DatabaseEngine(threading.Thread):
singleton = None
@ -17,9 +17,14 @@ class DatabaseEngine(threading.Thread):
self.event = threading.Event()
self.config = config
self.queue = queue
signal.signal(signal.SIGUSR1, self.signalHandler)
self.startTimer()
self.setDaemon(True)
def signalHandler(self, signum, frame):
Logger.log("SIGUSR1 received")
self.triggerStoring()
def triggerStoring(self):
self.event.set()

View File

@ -17,22 +17,42 @@ DEBUG_TO_STDOUT = True
BACKGROUND = False
PID_FILE = "/tmp/MqttDispatcher.pid"
LOG_FILE = "/tmp/MqttDispatcher.log"
MQTT_CONFIG = {
'host':'localhost',
'port':1883,
'tls':False,
'host':'eupenstrasse20.dynamic.hottis.de',
'port':8883,
'tls':True,
'ca':'ca.crt',
'user':'tron',
'password':'geheim123'
}
DATABASE_CONFIG = {
'period':300,
'host':'172.16.2.17',
'period':3600,
'host':'localhost',
'user':'smarthome',
'password':'smarthome123',
'db':'smarthome'
}
#MQTT_CONFIG = {
# 'host':'localhost',
# 'port':1883,
# 'tls':False,
# 'ca':'ca.crt',
# 'user':'tron',
# 'password':'geheim123'
#}
#DATABASE_CONFIG = {
# 'period':300,
# 'host':'172.16.2.17',
# 'user':'smarthome',
# 'password':'smarthome123',
# 'db':'smarthome'
#}
QUEUE_CONFIG = {
'file':'/tmp/mqttDispatcherQueue'
}
if BACKGROUND:
pid = os.fork()
@ -71,7 +91,7 @@ try:
mqttReader.registerParser(modbusParser)
Logger.log("ModbusParser started ...")
persistentQueue = PersistentQueue()
persistentQueue = PersistentQueue(QUEUE_CONFIG)
mbusParser.setNextStage(persistentQueue)
modbusParser.setNextStage(persistentQueue)
Logger.log("PersistentQueue instantiated ...")

View File

@ -1,17 +1,19 @@
from logger import Logger
import Queue
from persistqueue import Queue
class PersistentQueue(object):
def __init__(self):
def __init__(self, config):
super(PersistentQueue, self).__init__()
self.queue = Queue.Queue()
self.queue = Queue(config['file'])
def execute(self, data):
#Logger.log("PersistentQueue %s" % (str(data)))
Logger.log("PersistentQueue %s" % (str(data)))
self.queue.put_nowait(data)
def empty(self):
return self.queue.empty()
return not self.queue.qsize()
def get(self):
return self.queue.get()
item = self.queue.get()
self.queue.task_done()
return item