Use a single mqtt connection for publish and subscribe operations instead of separate one-shot connections for each update (#35)
This commit is contained in:
@@ -1,29 +1,25 @@
|
|||||||
import paho.mqtt.publish as publish
|
|
||||||
import paho.mqtt.client as mqtt
|
import paho.mqtt.client as mqtt
|
||||||
|
|
||||||
|
|
||||||
class MqttClient:
|
class MqttClient:
|
||||||
|
|
||||||
def __init__(self, config):
|
def __init__(self, config):
|
||||||
self._config = config
|
self._config = config
|
||||||
self._mqttc = mqtt.Client(client_id=self.client_id, clean_session=False)
|
self._mqttc = mqtt.Client(client_id=self.client_id, clean_session=False)
|
||||||
|
|
||||||
if self.username is not None and self.password is not None:
|
if self.username and self.password:
|
||||||
self._auth = {'username': self.username, 'password': self.password}
|
self.mqttc.username_pw_set(self.username, self.password)
|
||||||
else:
|
|
||||||
self._auth = None
|
self.mqttc.connect(self.hostname, port=self.port)
|
||||||
|
self.mqttc.loop_start()
|
||||||
|
|
||||||
def publish(self, messages):
|
def publish(self, messages):
|
||||||
if messages is None or len(messages) == 0:
|
if not messages:
|
||||||
return
|
return
|
||||||
|
|
||||||
if self.topic_prefix is not None:
|
|
||||||
for m in messages:
|
for m in messages:
|
||||||
m.topic = "{}/{}".format(self.topic_prefix, m.topic)
|
topic = "{}/{}".format(self.topic_prefix, m.topic) if self.topic_prefix else m.topic
|
||||||
|
self.mqttc.publish(topic, m.payload)
|
||||||
publish.multiple(list(map(lambda m: m.as_dict, messages)),
|
|
||||||
hostname=self.hostname,
|
|
||||||
port=self.port,
|
|
||||||
auth=self._auth)
|
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def client_id(self):
|
def client_id(self):
|
||||||
@@ -54,18 +50,11 @@ class MqttClient:
|
|||||||
return self._mqttc
|
return self._mqttc
|
||||||
|
|
||||||
def callbacks_subscription(self, callbacks):
|
def callbacks_subscription(self, callbacks):
|
||||||
if self.username is not None and self.password is not None:
|
|
||||||
self.mqttc.username_pw_set(self.username, self.password)
|
|
||||||
self.mqttc.connect(self.hostname, port=self.port)
|
|
||||||
|
|
||||||
for topic, callback in callbacks:
|
for topic, callback in callbacks:
|
||||||
if self.topic_prefix:
|
topic = "{}/{}".format(self.topic_prefix, topic) if self.topic_prefix else topic
|
||||||
topic = "{}/{}".format(self.topic_prefix, topic)
|
|
||||||
self.mqttc.message_callback_add(topic, callback)
|
self.mqttc.message_callback_add(topic, callback)
|
||||||
self.mqttc.subscribe(topic)
|
self.mqttc.subscribe(topic)
|
||||||
|
|
||||||
self.mqttc.loop_start()
|
|
||||||
|
|
||||||
|
|
||||||
class MqttMessage:
|
class MqttMessage:
|
||||||
def __init__(self, topic=None, payload=None):
|
def __init__(self, topic=None, payload=None):
|
||||||
|
|||||||
Reference in New Issue
Block a user