Added exceptioning handling
This commit is contained in:
parent
c11943cc67
commit
b599a4a483
1 changed files with 52 additions and 30 deletions
|
@ -33,40 +33,54 @@ def main():
|
||||||
if os.path.exists(socketFile):
|
if os.path.exists(socketFile):
|
||||||
os.remove(socketFile)
|
os.remove(socketFile)
|
||||||
|
|
||||||
stop_event = threading.Event()
|
stopEvent = threading.Event()
|
||||||
|
failureEvent = threading.Event()
|
||||||
|
|
||||||
dataBuffer = []
|
dataBuffer = []
|
||||||
|
|
||||||
def publishData(stop_event):
|
def publishData(stopEvent):
|
||||||
|
|
||||||
|
try:
|
||||||
|
|
||||||
if config.loggingMode == 'rabbitmq':
|
if config.loggingMode == 'rabbitmq':
|
||||||
|
|
||||||
import rabbitmq
|
import rabbitmq
|
||||||
rabbitmq = rabbitmq.rabbitMQClient(config.rabbitmqca,config.rabbitmqcacert,config.rabbitmqcakey,config.rabbitmqHost,config.rabbitmqPort,config.rabbitmqRoutingKey)
|
rabbitmq = rabbitmq.rabbitMQClient(config.rabbitmqca,config.rabbitmqcacert,config.rabbitmqcakey,config.rabbitmqHost,config.rabbitmqPort,config.rabbitmqRoutingKey)
|
||||||
|
|
||||||
while not (stop_event.is_set()):
|
while not (stopEvent.is_set() and not failureEvent.is_set()):
|
||||||
|
|
||||||
if dataBuffer:
|
if dataBuffer:
|
||||||
|
|
||||||
data = dataBuffer.pop(0)
|
data = dataBuffer.pop(0)
|
||||||
|
|
||||||
if config.loggingMode == 'rabbitmq':
|
if config.loggingMode == 'rabbitmq':
|
||||||
|
|
||||||
rabbitmq.publish(f"{data}")
|
rabbitmq.publish(f"{data}")
|
||||||
|
|
||||||
else:
|
else:
|
||||||
|
|
||||||
print("Not yet implemented")
|
print("Not yet implemented")
|
||||||
|
|
||||||
|
except Exception:
|
||||||
|
failureEvent.set()
|
||||||
|
|
||||||
def cleanup(signum, frame):
|
def cleanup(signum, frame):
|
||||||
|
|
||||||
print("Signal received, shutting down...")
|
print("Signal received, shutting down...")
|
||||||
stop_event.set()
|
stopEvent.set()
|
||||||
|
|
||||||
signal.signal(signal.SIGTERM, cleanup)
|
signal.signal(signal.SIGTERM, cleanup)
|
||||||
signal.signal(signal.SIGINT, cleanup)
|
signal.signal(signal.SIGINT, cleanup)
|
||||||
|
|
||||||
def server(stop_event):
|
def server(stopEvent):
|
||||||
|
|
||||||
|
try:
|
||||||
|
|
||||||
serverSocket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
serverSocket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
||||||
serverSocket.bind(socketFile)
|
serverSocket.bind(socketFile)
|
||||||
serverSocket.listen(50)
|
serverSocket.listen(50)
|
||||||
|
|
||||||
while not (stop_event.is_set()):
|
while not (stopEvent.is_set() and not failureEvent.is_set()):
|
||||||
|
|
||||||
readable, _, _ = select.select([serverSocket], [], [], 1.0)
|
readable, _, _ = select.select([serverSocket], [], [], 1.0)
|
||||||
|
|
||||||
|
@ -81,8 +95,12 @@ def main():
|
||||||
|
|
||||||
clientSocket.close()
|
clientSocket.close()
|
||||||
|
|
||||||
publishThread = threading.Thread(target=publishData, args=(stop_event,))
|
except Exception:
|
||||||
serverThread = threading.Thread(target=server, args=(stop_event,))
|
|
||||||
|
failureEvent.set()
|
||||||
|
|
||||||
|
publishThread = threading.Thread(target=publishData, args=(stopEvent,))
|
||||||
|
serverThread = threading.Thread(target=server, args=(stopEvent,))
|
||||||
|
|
||||||
publishThread.start()
|
publishThread.start()
|
||||||
serverThread.start()
|
serverThread.start()
|
||||||
|
@ -90,5 +108,9 @@ def main():
|
||||||
publishThread.join()
|
publishThread.join()
|
||||||
serverThread.join()
|
serverThread.join()
|
||||||
|
|
||||||
|
if failure_event.is_set():
|
||||||
|
print("One of the threads failed. Terminating")
|
||||||
|
sys.exit(1)
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
main()
|
main()
|
||||||
|
|
Loading…
Reference in a new issue