Mercurial > code > home > repos > mqtt_metrics

comparison mqtt_metrics.py @ 3:41e36f98f3b8

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
die on mqtt connection errors
author drewp@bigasterisk.com
date Fri, 09 Aug 2024 18:36:39 -0700
parents 579df3a4e62d
children cd1b8d7bda78
comparison
equal deleted inserted replaced
2:579df3a4e62d 3:41e36f98f3b8
1 import asyncio 1 import asyncio
2 import json 2 import json
3 import logging 3 import logging
4 import os
4 import re 5 import re
5 import time 6 import time
6 from typing import cast 7 from typing import cast
7 from weakref import WeakSet 8 from weakref import WeakSet
8 9
54 55
55 return message 56 return message
56 57
57 58
58 async def mqttTask(): 59 async def mqttTask():
59 client = aiomqtt.Client('mqtt2.bigasterisk.com', identifier="mqtt-exporter") 60 try:
60 async with client: 61 client = aiomqtt.Client('mqtt2.bigasterisk.com', identifier="mqtt-exporter")
61 await client.subscribe('#') 62 async with client:
62 async for mqttMessage in client.messages: 63 await client.subscribe('#')
63 onMqttMessage(mqttMessage) 64 async for mqttMessage in client.messages:
65 try:
66 onMqttMessage(mqttMessage)
67 except Exception as e:
68 log.warning("mqtt message (topic %r) failed: %r", mqttMessage.topic.value, e)
64 69
70 except Exception:
71 log.error("mqtt task failed", exc_info=True)
72 os.abort()
65 73
66 def onMqttMessage(mqttMessage): 74 def onMqttMessage(mqttMessage):
67 message = simplifyMqttMessage(mqttMessage) 75 message = simplifyMqttMessage(mqttMessage)
68 76
69 metricEvent = tryConverters(message) 77 metricEvent = tryConverters(message)