platypush/plugins/mqtt/__init__.py
def publish(self, topic: str, msg: Any, host: Optional[str] = None, port: Optional[int] = None,
                reply_topic: Optional[str] = None, timeout: int = 60,
                tls_cafile: Optional[str] = None, tls_certfile: Optional[str] = None,
                tls_keyfile: Optional[str] = None, tls_version: Optional[str] = None,
                tls_ciphers: Optional[str] = None, tls_insecure: Optional[bool] = None,
                username: Optional[str] = None, password: Optional[str] = None):
        """
        Sends a message to a topic.

        :param topic: Topic/channel where the message will be delivered
        :param msg: Message to be sent. It can be a list, a dict, or a Message object.
        :param host: MQTT broker hostname/IP (default: default host configured on the plugin).
        :param port: MQTT broker port (default: default port configured on the plugin).
        :param reply_topic: If a ``reply_topic`` is specified, then the action will wait for a response on this topic.
        :param timeout: If ``reply_topic`` is set, use this parameter to specify the maximum amount of time to
            wait for a response (default: 60 seconds).
        :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).
        :param tls_version: If TLS/SSL is enabled on the MQTT server and it requires a certain TLS version, specify it
            here (default: None). Supported versions: ``tls`` (automatic), ``tlsv1``, ``tlsv1.1``, ``tlsv1.2``.
        :param tls_insecure: Set to True to ignore TLS insecure warnings (default: False).
        :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).
        :param username: Specify it if the MQTT server requires authentication (default: None).
        :param password: Specify it if the MQTT server requires authentication (default: None).
        """
        response_buffer = io.BytesIO()
        client = None

        try:
            # Try to parse it as a platypush message or dump it to JSON from a dict/list
            if isinstance(msg, (dict, list)):
                msg = json.dumps(msg)

                try:
                    msg = Message.build(json.loads(msg))
                except Exception as e:
                    self.logger.debug(f'Not a valid JSON: {str(e)}')

            host = host or self.host
            port = port or self.port or 1883
            assert host, 'No host specified'

            client = self._get_client(tls_cafile=tls_cafile, tls_certfile=tls_certfile, tls_keyfile=tls_keyfile,
                                      tls_version=tls_version, tls_ciphers=tls_ciphers, tls_insecure=tls_insecure,
                                      username=username, password=password)

            client.connect(host, port, keepalive=timeout)
            response_received = threading.Event()

            if reply_topic:
                client.on_message = self._response_callback(reply_topic=reply_topic,
                                                            event=response_received,
                                                            buffer=response_buffer)
                client.subscribe(reply_topic)

            client.publish(topic, str(msg))
            if not reply_topic:
                return

            client.loop_start()
            ok = response_received.wait(timeout=timeout)
            if not ok:
                raise TimeoutError('Response timed out')
            return response_buffer.getvalue()
        finally:
            response_buffer.close()

            if client:
                try:
                    client.loop_stop()
                except Exception as e:
                    self.logger.warning(f'Could not stop client loop: {e}')

                client.disconnect()

        Sends a message to a topic.

        :param topic: Topic/channel where the message will be delivered
        :param msg: Message to be sent. It can be a list, a dict, or a Message object.
        :param host: MQTT broker hostname/IP (default: default host configured on the plugin).
        :param port: MQTT broker port (default: default port configured on the plugin).
        :param reply_topic: If a ``reply_topic`` is specified, then the action will wait for a response on this topic.
        :param timeout: If ``reply_topic`` is set, use this parameter to specify the maximum amount of time to
            wait for a response (default: 60 seconds).
        :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).
        :param tls_version: If TLS/SSL is enabled on the MQTT server and it requires a certain TLS version, specify it
            here (default: None). Supported versions: ``tls`` (automatic), ``tlsv1``, ``tlsv1.1``, ``tlsv1.2``.
        :param tls_insecure: Set to True to ignore TLS insecure warnings (default: False).
        :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).
        :param username: Specify it if the MQTT server requires authentication (default: None).
        :param password: Specify it if the MQTT server requires authentication (default: None).
        
Sends a message to a topic.
Sends a message to a topic .
def publish(self, topic: str, msg: Any, host: Optional[str] = None, port: Optional[int] = None,
                reply_topic: Optional[str] = None, timeout: int = 60,
                tls_cafile: Optional[str] = None, tls_certfile: Optional[str] = None,
                tls_keyfile: Optional[str] = None, tls_version: Optional[str] = None,
                tls_ciphers: Optional[str] = None, tls_insecure: Optional[bool] = None,
                username: Optional[str] = None, password: Optional[str] = None):
        response_buffer = io.BytesIO()
        client = None
        try:
            if isinstance(msg, (dict, list)):
                msg = json.dumps(msg)
                try:
                    msg = Message.build(json.loads(msg))
                except Exception as e:
                    self.logger.debug(f'Not a valid JSON: {str(e)}')
            host = host or self.host
            port = port or self.port or 1883
            assert host, 'No host specified'
            client = self._get_client(tls_cafile=tls_cafile, tls_certfile=tls_certfile, tls_keyfile=tls_keyfile,
                                      tls_version=tls_version, tls_ciphers=tls_ciphers, tls_insecure=tls_insecure,
                                      username=username, password=password)
            client.connect(host, port, keepalive=timeout)
            response_received = threading.Event()
            if reply_topic:
                client.on_message = self._response_callback(reply_topic=reply_topic,
                                                            event=response_received,
                                                            buffer=response_buffer)
                client.subscribe(reply_topic)
            client.publish(topic, str(msg))
            if not reply_topic:
                return
            client.loop_start()
            ok = response_received.wait(timeout=timeout)
            if not ok:
                raise TimeoutError('Response timed out')
            return response_buffer.getvalue()
        finally:
            response_buffer.close()
            if client:
                try:
                    client.loop_stop()
                except Exception as e:
                    self.logger.warning(f'Could not stop client loop: {e}')
                client.disconnect()
def publish ( self , topic : str , msg : Any , host : Optional [ str ] = None , port : Optional [ int ] = None , reply_topic : Optional [ str ] = None , timeout : int = 60 , tls_cafile : Optional [ str ] = None , tls_certfile : Optional [ str ] = None , tls_keyfile : Optional [ str ] = None , tls_version : Optional [ str ] = None , tls_ciphers : Optional [ str ] = None , tls_insecure : Optional [ bool ] = None , username : Optional [ str ] = None , password : Optional [ str ] = None ) : response_buffer = io . BytesIO ( ) client = None try : if isinstance ( msg , ( dict , list ) ) : msg = json . dumps ( msg ) try : msg = Message . build ( json . loads ( msg ) ) except Exception as e : self . logger . debug ( f'Not a valid JSON: {str(e)}' ) host = host or self . host port = port or self . port or 1883 assert host , 'No host specified' client = self . _get_client ( tls_cafile = tls_cafile , tls_certfile = tls_certfile , tls_keyfile = tls_keyfile , tls_version = tls_version , tls_ciphers = tls_ciphers , tls_insecure = tls_insecure , username = username , password = password ) client . connect ( host , port , keepalive = timeout ) response_received = threading . Event ( ) if reply_topic : client . on_message = self . _response_callback ( reply_topic = reply_topic , event = response_received , buffer = response_buffer ) client . subscribe ( reply_topic ) client . publish ( topic , str ( msg ) ) if not reply_topic : return client . loop_start ( ) ok = response_received . wait ( timeout = timeout ) if not ok : raise TimeoutError ( 'Response timed out' ) return response_buffer . getvalue ( ) finally : response_buffer . close ( ) if client : try : client . loop_stop ( ) except Exception as e : self . logger . warning ( f'Could not stop client loop: {e}' ) client . disconnect ( )
Sends a message to a topic.
Sends a message to a topic .
"""
        Sends a message to a topic.

        :param topic: Topic/channel where the message will be delivered
        :param msg: Message to be sent. It can be a list, a dict, or a Message object.
        :param host: MQTT broker hostname/IP (default: default host configured on the plugin).
        :param port: MQTT broker port (default: default port configured on the plugin).
        :param reply_topic: If a ``reply_topic`` is specified, then the action will wait for a response on this topic.
        :param timeout: If ``reply_topic`` is set, use this parameter to specify the maximum amount of time to
            wait for a response (default: 60 seconds).
        :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).
        :param tls_version: If TLS/SSL is enabled on the MQTT server and it requires a certain TLS version, specify it
            here (default: None). Supported versions: ``tls`` (automatic), ``tlsv1``, ``tlsv1.1``, ``tlsv1.2``.
        :param tls_insecure: Set to True to ignore TLS insecure warnings (default: False).
        :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).
        :param username: Specify it if the MQTT server requires authentication (default: None).
        :param password: Specify it if the MQTT server requires authentication (default: None).
        """ # Try to parse it as a platypush message or dump it to JSON from a dict/list
self
topic
str
Topic/channel where the message will be delivered
Topic / channel where the message will be delivered
msg
Any
Message to be sent. It can be a list, a dict, or a Message object.
Message to be sent . It can be a list a dict or a Message object .
host
Optional[str]
MQTT broker hostname/IP (default: default host configured on the plugin).
MQTT broker hostname / IP ( default : default host configured on the plugin ) .
port
Optional[int]
MQTT broker port (default: default port configured on the plugin).
MQTT broker port ( default : default port configured on the plugin ) .
reply_topic
Optional[str]
If a ``reply_topic`` is specified, then the action will wait for a response on this topic.
If a ` ` reply_topic ` ` is specified then the action will wait for a response on this topic .
timeout
int
If ``reply_topic`` is set, use this parameter to specify the maximum amount of time to
wait for a response (default: 60 seconds).
If ` ` reply_topic ` ` is set use this parameter to specify the maximum amount of time to wait for a response ( default : 60 seconds ) .
tls_cafile
Optional[str]
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).
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 ) .
tls_certfile
Optional[str]
If TLS/SSL is enabled on the MQTT server and a client certificate it required, specify it
here (default: None).
If TLS / SSL is enabled on the MQTT server and a client certificate it required specify it here ( default : None ) .
tls_keyfile
Optional[str]
If TLS/SSL is enabled on the MQTT server and a client certificate key it required, specify
it here (default: None).
If TLS / SSL is enabled on the MQTT server and a client certificate key it required specify it here ( default : None ) .
tls_version
Optional[str]
If TLS/SSL is enabled on the MQTT server and it requires a certain TLS version, specify it
here (default: None).
If TLS / SSL is enabled on the MQTT server and it requires a certain TLS version specify it here ( default : None ) .
tls_ciphers
Optional[str]
If TLS/SSL is enabled on the MQTT server and an explicit list of supported ciphers is
required, specify it here (default: None).
If TLS / SSL is enabled on the MQTT server and an explicit list of supported ciphers is required specify it here ( default : None ) .
tls_insecure
Optional[bool]
Set to True to ignore TLS insecure warnings (default: False).
Set to True to ignore TLS insecure warnings ( default : False ) .
username
Optional[str]
Specify it if the MQTT server requires authentication (default: None).
Specify it if the MQTT server requires authentication ( default : None ) .
password
Optional[str]
Specify it if the MQTT server requires authentication (default: None).
Specify it if the MQTT server requires authentication ( default : None ) .