Add Service and MqttSubscriber classes
- Create utils dir and move utils.py there
This commit is contained in:
51
motor_passo/utils/mqtt_subscriber.py
Normal file
51
motor_passo/utils/mqtt_subscriber.py
Normal file
@@ -0,0 +1,51 @@
|
||||
import paho.mqtt.client as mqtt
|
||||
|
||||
from motor_passo.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}")
|
||||
Reference in New Issue
Block a user