2018-05-27 12:21:37 +02:00
|
|
|
import json
|
2018-11-02 16:15:48 +01:00
|
|
|
import os
|
2018-10-26 15:33:23 +02:00
|
|
|
import threading
|
2020-01-22 18:34:28 +01:00
|
|
|
from typing import Optional
|
2018-05-27 12:21:37 +02:00
|
|
|
|
|
|
|
from platypush.backend import Backend
|
2020-08-27 16:41:51 +02:00
|
|
|
from platypush.config import Config
|
2018-11-02 16:15:48 +01:00
|
|
|
from platypush.context import get_plugin
|
2018-05-27 12:21:37 +02:00
|
|
|
from platypush.message import Message
|
2019-03-07 22:51:58 +01:00
|
|
|
from platypush.message.event.mqtt import MQTTMessageEvent
|
2018-10-26 15:33:23 +02:00
|
|
|
from platypush.message.request import Request
|
2020-08-27 12:44:00 +02:00
|
|
|
from platypush.plugins.mqtt import MqttPlugin as MQTTPlugin
|
2019-01-10 23:45:13 +01:00
|
|
|
from platypush.utils import set_thread_name
|
2018-05-27 12:21:37 +02:00
|
|
|
|
|
|
|
|
|
|
|
class MqttBackend(Backend):
|
|
|
|
"""
|
2018-06-26 00:16:39 +02:00
|
|
|
Backend that reads messages from a configured MQTT topic (default:
|
|
|
|
``platypush_bus_mq/<device_id>``) and posts them to the application bus.
|
|
|
|
|
2019-03-07 22:51:58 +01:00
|
|
|
Triggers:
|
|
|
|
|
|
|
|
* :class:`platypush.message.event.mqtt.MQTTMessageEvent` when a new
|
|
|
|
message is received on one of the custom listeners
|
|
|
|
|
2018-06-26 00:16:39 +02:00
|
|
|
Requires:
|
|
|
|
|
|
|
|
* **paho-mqtt** (``pip install paho-mqtt``)
|
2018-05-27 12:21:37 +02:00
|
|
|
"""
|
|
|
|
|
2019-03-07 22:51:58 +01:00
|
|
|
_default_mqtt_port = 1883
|
|
|
|
|
2020-01-22 18:34:28 +01:00
|
|
|
def __init__(self, host: Optional[str] = None, port: int = _default_mqtt_port,
|
|
|
|
topic='platypush_bus_mq', subscribe_default_topic: bool = True,
|
|
|
|
tls_cafile: Optional[str] = None, tls_certfile: Optional[str] = None,
|
|
|
|
tls_keyfile: Optional[str] = None, tls_version: Optional[str] = None,
|
2020-08-27 15:56:43 +02:00
|
|
|
tls_ciphers: Optional[str] = None, tls_insecure: bool = False,
|
2020-08-27 16:41:51 +02:00
|
|
|
username: Optional[str] = None, password: Optional[str] = None,
|
|
|
|
client_id: Optional[str] = None, listeners=None,
|
2020-08-27 15:56:43 +02:00
|
|
|
*args, **kwargs):
|
2018-05-27 12:21:37 +02:00
|
|
|
"""
|
2018-06-26 00:16:39 +02:00
|
|
|
:param host: MQTT broker host
|
|
|
|
:param port: MQTT broker port (default: 1883)
|
|
|
|
:param topic: Topic to read messages from (default: ``platypush_bus_mq/<device_id>``)
|
2020-01-22 18:34:28 +01:00
|
|
|
:param subscribe_default_topic: Whether the backend should subscribe the default topic (default:
|
|
|
|
``platypush_bus_mq/<device_id>``) and execute the messages received there as action requests
|
|
|
|
(default: True).
|
2019-12-08 23:46:34 +01:00
|
|
|
:param tls_cafile: If TLS/SSL is enabled on the MQTT server and the certificate requires a certificate authority
|
|
|
|
to authenticate it, `ssl_cafile` will point to the provided ca.crt file (default: None)
|
|
|
|
:param tls_certfile: If TLS/SSL is enabled on the MQTT server and a client certificate it required, specify it
|
|
|
|
here (default: None)
|
|
|
|
:param tls_keyfile: If TLS/SSL is enabled on the MQTT server and a client certificate key it required,
|
|
|
|
specify it here (default: None) :type tls_keyfile: str
|
|
|
|
:param tls_version: If TLS/SSL is enabled on the MQTT server and it requires a certain TLS version, specify it
|
2020-08-27 12:44:00 +02:00
|
|
|
here (default: None). Supported versions: ``tls`` (automatic), ``tlsv1``, ``tlsv1.1``, ``tlsv1.2``.
|
2019-12-08 23:46:34 +01:00
|
|
|
:param tls_ciphers: If TLS/SSL is enabled on the MQTT server and an explicit list of supported ciphers is
|
|
|
|
required, specify it here (default: None)
|
2020-08-27 15:56:43 +02:00
|
|
|
:param tls_insecure: Set to True to ignore TLS insecure warnings (default: False).
|
2018-11-02 16:15:48 +01:00
|
|
|
:param username: Specify it if the MQTT server requires authentication (default: None)
|
|
|
|
:param password: Specify it if the MQTT server requires authentication (default: None)
|
2020-08-27 16:41:51 +02:00
|
|
|
:param client_id: ID used to identify the client on the MQTT server (default: None).
|
|
|
|
If None is specified then ``Config.get('device_id')`` will be used.
|
2019-03-07 22:51:58 +01:00
|
|
|
:param listeners: If specified then the MQTT backend will also listen for
|
|
|
|
messages on the additional configured message queues. This parameter
|
|
|
|
is a list of maps where each item supports the same arguments passed
|
|
|
|
to the main backend configuration (host, port, topic, password etc.).
|
|
|
|
Note that the message queue configured on the main configuration
|
|
|
|
will expect valid Platypush messages that then can execute, while
|
2020-01-22 18:34:28 +01:00
|
|
|
message queues registered to the listeners will accept any message. Example::
|
|
|
|
|
|
|
|
listeners:
|
|
|
|
- host: localhost
|
|
|
|
topics:
|
|
|
|
- topic1
|
|
|
|
- topic2
|
|
|
|
- topic3
|
|
|
|
- host: sensors
|
|
|
|
topics:
|
|
|
|
- topic4
|
|
|
|
- topic5
|
|
|
|
|
2018-05-27 12:21:37 +02:00
|
|
|
"""
|
2018-06-26 00:16:39 +02:00
|
|
|
|
2018-05-27 12:21:37 +02:00
|
|
|
super().__init__(*args, **kwargs)
|
|
|
|
|
|
|
|
self.host = host
|
|
|
|
self.port = port
|
|
|
|
self.topic = '{}/{}'.format(topic, self.device_id)
|
2020-01-22 18:34:28 +01:00
|
|
|
self.subscribe_default_topic = subscribe_default_topic
|
2018-11-02 16:15:48 +01:00
|
|
|
self.username = username
|
|
|
|
self.password = password
|
2020-08-27 16:41:51 +02:00
|
|
|
self.client_id = client_id or Config.get('device_id')
|
2019-01-18 04:10:27 +01:00
|
|
|
self._client = None
|
2019-03-07 22:51:58 +01:00
|
|
|
self._listeners = []
|
2018-05-27 12:21:37 +02:00
|
|
|
|
2020-08-27 16:29:21 +02:00
|
|
|
self.tls_cafile = self._expandpath(tls_cafile) if tls_cafile else None
|
|
|
|
self.tls_certfile = self._expandpath(tls_certfile) if tls_certfile else None
|
|
|
|
self.tls_keyfile = self._expandpath(tls_keyfile) if tls_keyfile else None
|
2020-08-27 12:44:00 +02:00
|
|
|
self.tls_version = MQTTPlugin.get_tls_version(tls_version)
|
2018-11-02 16:15:48 +01:00
|
|
|
self.tls_ciphers = tls_ciphers
|
2020-08-27 15:56:43 +02:00
|
|
|
self.tls_insecure = tls_insecure
|
2019-03-07 22:51:58 +01:00
|
|
|
self.listeners_conf = listeners or []
|
2018-11-02 16:15:48 +01:00
|
|
|
|
2020-01-22 18:34:28 +01:00
|
|
|
def send_message(self, msg, topic: Optional[str] = None, **kwargs):
|
2018-11-06 11:40:01 +01:00
|
|
|
try:
|
|
|
|
client = get_plugin('mqtt')
|
2020-01-22 18:34:28 +01:00
|
|
|
client.send_message(topic=topic or self.topic, msg=msg, host=self.host,
|
2018-11-06 11:40:01 +01:00
|
|
|
port=self.port, username=self.username,
|
|
|
|
password=self.password, tls_cafile=self.tls_cafile,
|
2020-08-27 15:56:43 +02:00
|
|
|
tls_certfile=self.tls_certfile, tls_keyfile=self.tls_keyfile,
|
|
|
|
tls_version=self.tls_version, tls_insecure=self.tls_insecure,
|
2020-08-27 16:41:51 +02:00
|
|
|
tls_ciphers=self.tls_ciphers, client_id=self.client_id, **kwargs)
|
2018-11-06 11:40:01 +01:00
|
|
|
except Exception as e:
|
|
|
|
self.logger.exception(e)
|
2018-05-27 12:21:37 +02:00
|
|
|
|
2020-01-22 18:34:28 +01:00
|
|
|
@staticmethod
|
|
|
|
def on_connect(*topics):
|
|
|
|
# noinspection PyUnusedLocal
|
|
|
|
def handler(client, userdata, flags, rc):
|
|
|
|
for topic in topics:
|
|
|
|
client.subscribe(topic)
|
2019-12-08 23:46:34 +01:00
|
|
|
|
2020-01-22 18:34:28 +01:00
|
|
|
return handler
|
2019-03-07 22:51:58 +01:00
|
|
|
|
2020-01-22 18:34:28 +01:00
|
|
|
def on_mqtt_message(self):
|
|
|
|
def handler(client, _, msg):
|
2019-03-07 22:51:58 +01:00
|
|
|
data = msg.payload
|
2019-12-08 23:46:34 +01:00
|
|
|
# noinspection PyBroadException
|
2019-03-07 22:51:58 +01:00
|
|
|
try:
|
|
|
|
data = data.decode('utf-8')
|
|
|
|
data = json.loads(data)
|
|
|
|
except:
|
|
|
|
pass
|
|
|
|
|
2019-12-08 23:46:34 +01:00
|
|
|
# noinspection PyProtectedMember
|
2020-01-22 18:34:28 +01:00
|
|
|
self.bus.post(MQTTMessageEvent(host=client._host, port=client._port, topic=msg.topic, msg=data))
|
|
|
|
|
|
|
|
return handler
|
|
|
|
|
2020-08-27 16:29:21 +02:00
|
|
|
@staticmethod
|
2020-08-27 16:30:18 +02:00
|
|
|
def _expandpath(path: str) -> str:
|
2020-08-27 16:29:21 +02:00
|
|
|
return os.path.abspath(os.path.expanduser(path)) if path else path
|
|
|
|
|
2020-01-22 18:34:28 +01:00
|
|
|
def _initialize_listeners(self, listeners_conf):
|
|
|
|
import paho.mqtt.client as mqtt
|
2019-03-07 22:51:58 +01:00
|
|
|
|
2020-01-22 18:34:28 +01:00
|
|
|
# noinspection PyShadowingNames
|
|
|
|
def listener_thread(client, host, port):
|
|
|
|
client.connect(host, port)
|
|
|
|
client.loop_forever()
|
|
|
|
|
|
|
|
# noinspection PyShadowingNames,PyUnusedLocal
|
2019-03-07 22:51:58 +01:00
|
|
|
for i, listener in enumerate(listeners_conf):
|
|
|
|
host = listener.get('host')
|
|
|
|
port = listener.get('port', self._default_mqtt_port)
|
2019-03-07 23:03:12 +01:00
|
|
|
topics = listener.get('topics')
|
2019-03-07 22:51:58 +01:00
|
|
|
username = listener.get('username')
|
|
|
|
password = listener.get('password')
|
2020-08-27 16:41:51 +02:00
|
|
|
client_id = listener.get('client_id', self.client_id)
|
2020-08-27 16:29:21 +02:00
|
|
|
tls_cafile = self._expandpath(listener.get('tls_cafile'))
|
2019-03-07 22:51:58 +01:00
|
|
|
|
2019-03-07 23:03:12 +01:00
|
|
|
if not host or not topics:
|
|
|
|
self.logger.warning('No host nor list of topics specified for ' +
|
2019-12-08 23:46:34 +01:00
|
|
|
'listener n.{}'.format(i + 1))
|
2019-03-07 22:51:58 +01:00
|
|
|
continue
|
|
|
|
|
2020-08-27 16:41:51 +02:00
|
|
|
client = mqtt.Client(client_id)
|
2020-01-22 18:34:28 +01:00
|
|
|
client.on_connect = self.on_connect(*topics)
|
|
|
|
client.on_message = self.on_mqtt_message()
|
2019-03-07 22:51:58 +01:00
|
|
|
|
|
|
|
if username and password:
|
|
|
|
client.username_pw_set(username, password)
|
|
|
|
|
|
|
|
if tls_cafile:
|
|
|
|
client.tls_set(ca_certs=tls_cafile,
|
2020-08-27 16:29:21 +02:00
|
|
|
certfile=self._expandpath(listener.get('tls_certfile')),
|
|
|
|
keyfile=self._expandpath(listener.get('tls_keyfile')),
|
2020-08-27 12:44:00 +02:00
|
|
|
tls_version=MQTTPlugin.get_tls_version(listener.get('tls_version')),
|
2019-03-07 22:51:58 +01:00
|
|
|
ciphers=listener.get('tls_ciphers'))
|
|
|
|
|
2020-08-27 15:56:43 +02:00
|
|
|
client.tls_insecure_set(self.tls_insecure)
|
|
|
|
|
2019-12-08 23:46:34 +01:00
|
|
|
threading.Thread(target=listener_thread, kwargs={
|
|
|
|
'client': client, 'host': host, 'port': port}).start()
|
2019-03-07 22:51:58 +01:00
|
|
|
|
2020-01-22 18:34:28 +01:00
|
|
|
def on_exec_message(self):
|
|
|
|
def handler(_, __, msg):
|
2019-12-08 23:46:34 +01:00
|
|
|
# noinspection PyShadowingNames
|
2018-10-26 15:33:23 +02:00
|
|
|
def response_thread(msg):
|
2019-01-13 20:41:15 +01:00
|
|
|
set_thread_name('MQTTProcessor')
|
2018-10-26 15:33:23 +02:00
|
|
|
response = self.get_message_response(msg)
|
2019-01-02 09:29:27 +01:00
|
|
|
if not response:
|
|
|
|
return
|
2018-10-26 15:33:23 +02:00
|
|
|
response_topic = '{}/responses/{}'.format(self.topic, msg.id)
|
|
|
|
|
|
|
|
self.logger.info('Processing response on the MQTT topic {}: {}'.
|
2019-12-08 23:46:34 +01:00
|
|
|
format(response_topic, response))
|
2018-10-26 15:33:23 +02:00
|
|
|
|
2018-11-06 11:40:01 +01:00
|
|
|
self.send_message(response)
|
2018-10-26 15:33:23 +02:00
|
|
|
|
2018-05-27 12:21:37 +02:00
|
|
|
msg = msg.payload.decode('utf-8')
|
2019-12-08 23:46:34 +01:00
|
|
|
# noinspection PyBroadException
|
|
|
|
try:
|
2020-01-22 18:34:28 +01:00
|
|
|
msg = json.loads(msg)
|
|
|
|
msg = Message.build(msg)
|
2019-12-08 23:46:34 +01:00
|
|
|
except:
|
|
|
|
pass
|
|
|
|
|
|
|
|
if not msg:
|
|
|
|
return
|
2018-05-27 12:21:37 +02:00
|
|
|
|
2018-06-06 20:09:18 +02:00
|
|
|
self.logger.info('Received message on the MQTT backend: {}'.format(msg))
|
2018-11-06 11:40:01 +01:00
|
|
|
|
|
|
|
try:
|
|
|
|
self.on_message(msg)
|
|
|
|
except Exception as e:
|
|
|
|
self.logger.exception(e)
|
2018-11-10 01:16:32 +01:00
|
|
|
return
|
2018-05-27 12:21:37 +02:00
|
|
|
|
2018-10-26 15:33:23 +02:00
|
|
|
if isinstance(msg, Request):
|
2020-01-22 18:34:28 +01:00
|
|
|
threading.Thread(target=response_thread, name='MQTTProcessor', args=(msg,)).start()
|
|
|
|
|
|
|
|
return handler
|
2018-10-26 15:33:23 +02:00
|
|
|
|
2020-01-22 18:34:28 +01:00
|
|
|
def run(self):
|
2019-01-18 04:10:27 +01:00
|
|
|
import paho.mqtt.client as mqtt
|
|
|
|
|
2018-10-25 20:45:58 +02:00
|
|
|
super().run()
|
2020-01-22 18:34:28 +01:00
|
|
|
self._client = None
|
|
|
|
|
|
|
|
if self.host:
|
2020-08-27 16:41:51 +02:00
|
|
|
self._client = mqtt.Client(self.client_id)
|
2020-01-22 18:34:28 +01:00
|
|
|
if self.subscribe_default_topic:
|
|
|
|
self._client.on_connect = self.on_connect(self.topic)
|
2018-11-02 16:15:48 +01:00
|
|
|
|
2020-01-22 18:34:28 +01:00
|
|
|
self._client.on_message = self.on_exec_message()
|
|
|
|
if self.username and self.password:
|
|
|
|
self._client.username_pw_set(self.username, self.password)
|
2018-11-02 16:15:48 +01:00
|
|
|
|
2020-01-22 18:34:28 +01:00
|
|
|
if self.tls_cafile:
|
|
|
|
self._client.tls_set(ca_certs=self.tls_cafile, certfile=self.tls_certfile,
|
2020-08-27 12:44:00 +02:00
|
|
|
keyfile=self.tls_keyfile,
|
|
|
|
tls_version=self.tls_version,
|
2020-01-22 18:34:28 +01:00
|
|
|
ciphers=self.tls_ciphers)
|
2018-11-02 16:15:48 +01:00
|
|
|
|
2020-08-27 15:56:43 +02:00
|
|
|
self._client.tls_insecure_set(self.tls_insecure)
|
|
|
|
|
2020-01-22 18:34:28 +01:00
|
|
|
self._client.connect(self.host, self.port, 60)
|
|
|
|
self.logger.info('Initialized MQTT backend on host {}:{}, topic {}'.
|
|
|
|
format(self.host, self.port, self.topic))
|
2018-05-27 12:21:37 +02:00
|
|
|
|
2019-03-07 22:51:58 +01:00
|
|
|
self._initialize_listeners(self.listeners_conf)
|
2020-01-22 18:34:28 +01:00
|
|
|
if self._client:
|
|
|
|
self._client.loop_forever()
|
2018-05-27 12:21:37 +02:00
|
|
|
|
2019-01-18 04:10:27 +01:00
|
|
|
def stop(self):
|
|
|
|
self.logger.info('Received STOP event on MqttBackend')
|
|
|
|
if self._client:
|
|
|
|
self._client.disconnect()
|
|
|
|
self._client.loop_stop()
|
|
|
|
self._client = None
|
2018-05-27 12:21:37 +02:00
|
|
|
|
2019-03-07 22:51:58 +01:00
|
|
|
for listener in self._listeners:
|
|
|
|
try:
|
|
|
|
listener.loop_stop()
|
|
|
|
except Exception as e:
|
2019-12-08 23:46:34 +01:00
|
|
|
# noinspection PyProtectedMember
|
2020-01-22 18:34:28 +01:00
|
|
|
self.logger.warning('Could not stop listener {host}:{port}: {error}'.format(
|
|
|
|
host=listener._host, port=listener._port,
|
|
|
|
error=str(e)))
|
|
|
|
|
2019-03-07 22:51:58 +01:00
|
|
|
|
2018-05-27 12:21:37 +02:00
|
|
|
# vim:sw=4:ts=4:et:
|