openrvdas.logger.readers.mqtt_reader

  1#!/usr/bin/env python3
  2
  3import logging
  4from queue import Queue
  5
  6# Don't barf if they don't have redis installed. Only complain if
  7# they actually try to use it, below
  8try:
  9    import paho.mqtt.client as mqtt  # import the client | $ pip installing paho-mqtt is necessary
 10    PAHO_ENABLED = True
 11
 12    # Check which paho version is being used
 13    from pkg_resources import get_distribution, packaging  # noqa: E402
 14    PAHO_VERSION = get_distribution('paho-mqtt').version
 15    if packaging.version.parse(PAHO_VERSION) >= packaging.version.parse('2.0.0'):
 16        USE_VERSION_FLAG = True
 17    else:
 18        USE_VERSION_FLAG = False
 19except ModuleNotFoundError:
 20    PAHO_ENABLED = False
 21
 22from logger.readers.reader import Reader  # noqa: E402
 23
 24
 25################################################################################
 26class MQTTReader(Reader):
 27    """
 28    Read messages from an mqtt broker
 29    """
 30
 31    def __init__(self, broker, channel, client_name,
 32                 port=1883, clean_start=None,
 33                 qos=0, return_as_bytes=False, **kwargs):
 34        """
 35        Read text records from the channel subscription.
 36        ```
 37
 38        broker       MQTT broker to connect, broker format[###.###.#.#]
 39        channel      MQTT channel to read from, channel format[@broker/path_of_subscripton]
 40        port         broker port, typically 1883
 41        clean_start  Request new session on first connection. Options: True, False,
 42                       or the default of mqtt.MQTT_CLEAN_START_FIRST_ONLY
 43        qos          Quality of service: 0 = at most once, 1 = at least once, 2 = exactly once
 44        return_as_bytes
 45                     If true, return message in bytes, otherwise convert to str
 46        ```
 47        Instructions on how to start an MQTT broker:
 48
 49        1. First install the Mosquitto Broker :
 50            ```
 51            sudo apt-get update
 52            sudo apt-get install mosquitto
 53            sudo apt-get install mosquitto-clients
 54            ```
 55        2. The mosquitto service starts automatically when downloaded but use :
 56            ```
 57            sudo service mosquitto start
 58            sudo service mosquitto stop
 59            ```
 60            to start and stop the service.
 61
 62        3. To test the install use:
 63            ```
 64            netstat -at
 65            ```
 66            and you should see the MQTT broker which is the port 1883
 67
 68        4. In order to manually subscribe to a client use :
 69            ```
 70            mosquitto_sub -t "example/topic"
 71            ```
 72            and publish a message by using
 73            ```
 74            mosquitto_pub -m "published message" -t "certain/channel"
 75            ```
 76        5. Mosquitto uses a configuration file "mosquitto.conf" which you can
 77           find in /etc/mosquitto 	folder
 78
 79        ```
 80        """
 81        if not PAHO_ENABLED:
 82            raise ModuleNotFoundError('MQTTReader(): paho-mqtt is not installed. Please '
 83                                      'try "pip install paho-mqtt" prior to use.')
 84        if qos not in [0, 1, 2]:
 85            raise ValueError('MQTTReader parameter qos must be integer value 0, 1 or 2. '
 86                             f'Found type "{type(qos).__name__}", value "{qos}".')
 87
 88        # Let's build it!
 89        super().__init__(**kwargs)
 90
 91        def on_connect(client, userdata, flags, rc, properties=None):
 92            logging.info(f'Connected With Result Code: {rc}')
 93
 94        def on_message(client, userdata, message):
 95            self.queue.put(message)
 96
 97        self.broker = broker
 98        self.channel = channel
 99        self.client_name = client_name
100        self.port = port
101        if clean_start is None:
102            clean_start = mqtt.MQTT_CLEAN_START_FIRST_ONLY
103        self.clean_start = clean_start
104        self.qos = qos
105        self.return_as_bytes = return_as_bytes
106        self.queue = Queue()
107
108        try:
109            if USE_VERSION_FLAG:
110                self.client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_name)
111            else:
112                self.client = mqtt.Client(client_name)
113
114            self.client.on_connect = on_connect
115            self.client.on_message = on_message
116
117            if USE_VERSION_FLAG:
118                self.client.connect(broker, port)
119                self.client.subscribe(channel, qos=self.qos)
120            else:
121                self.client.connect(broker, port, clean_start=clean_start)
122                self.client.subscribe(channel, options=SubscribeOptions(qos=qos))  # noqa: F821
123
124        except (mqtt.WebsocketConnectionError, ConnectionRefusedError) as e:
125            logging.error(f'Unable to connect to broker at {broker}:{port} {channel}')
126            raise e
127
128    ############################
129    def read(self):
130        while True:
131            try:
132                self.client.loop()
133                while not self.queue.empty():
134                    message = self.queue.get()
135                    if message is None:
136                        continue
137                    logging.debug('Got message "%s"', message.payload)
138                    if self.return_as_bytes:
139                        return message.payload
140                    else:
141                        return str(message.payload, 'utf-8')
142            except KeyboardInterrupt:
143                self.client.disconnect()
144                exit(0)