52 lines
1.7 KiB
Python
52 lines
1.7 KiB
Python
import paho.mqtt.client as mqtt
|
|
|
|
from spectrometer.utils.service import Service
|
|
|
|
|
|
class MqttSubscriber(Service):
|
|
|
|
def __init__(self, broker_host, broker_port):
|
|
super().__init__(name="mqtt_listener")
|
|
self.broker_host = broker_host
|
|
self.broker_port = broker_port
|
|
self._client = mqtt.Client()
|
|
self._client.on_connect = self._on_connect
|
|
self._client.on_message = self._on_message
|
|
self._client.on_disconnect = self._on_disconnect
|
|
self._cbs: dict[str, callable] = {}
|
|
|
|
def run(self) -> None:
|
|
self.client.connect(self.broker_host, self.broker_port, 60)
|
|
self.client.loop_start()
|
|
|
|
def subscribe(self, topic: str, cb: callable) -> None:
|
|
self._cbs[topic] = cb
|
|
|
|
if self._client.is_connected():
|
|
self._client.subscribe(topic)
|
|
|
|
def _on_connect(self, client, userdata, flags, rc) -> None:
|
|
self._l.debug(f"Conectado com resultado {str(rc)}")
|
|
|
|
for topic in self._cbs.keys():
|
|
client.subscribe(topic)
|
|
|
|
def _on_disconnect(self, client, userdata, rc) -> None:
|
|
self._l.debug(f"Desconectado, código {str(rc)}")
|
|
# Start the loop again in case of an unexpected disconnection
|
|
client.loop_start()
|
|
|
|
def _on_message(self, client, userdata, message) -> None:
|
|
payload = message.payload.decode("utf-8")
|
|
self._l.debug(f"Mensagem recebida no tópico: {message.topic}")
|
|
|
|
if message.topic not in self._cbs:
|
|
self._l.debug(f"Mensagem no tópico {message.topic} não tratada")
|
|
return
|
|
|
|
try:
|
|
self._cbs[message.topic](message.topic, payload)
|
|
except Exception:
|
|
self._l.exception(
|
|
f"Falha ao processar mensagem no tópico {message.topic}")
|