Introduction to MQTT (Introduction)

■ What is mosquitto?

Open source software that implements the MQTT protocol. http://mosquitto.org/

It seems that there is also Apache Apollo as another OSS that implements MQTT.

● What is MQTT?

Publish / subscribe (publish-subscribe) model protocol https://sango.shiguredo.jp/mqtt

A type of asynchronous messaging paradigm, a program that allows a message sender (publisher) to send a message without assuming a specific recipient (subscriber). https://ja.wikipedia.org/wiki/%E5%87%BA%E7%89%88-%E8%B3%BC%E8%AA%AD%E5%9E%8B%E3%83%A2%E3%83%87%E3%83%AB

● Asynchronous message

In asynchronous messaging, messages are thrown one after another without waiting for the result, so the timing of processing the messages does not match. http://ledsun.hatenablog.com/entry/2013/07/18/181044

■ Introduction of mosquitto

● Installation of mosquitto

The following site will be helpful. http://dev.classmethod.jp/cloud/setting-up-mosquitto-logging-on-amazon-linux/

[root@localhost tmp]# wget http://download.opensuse.org/repositories/home:/oojah:/mqtt/CentOS_CentOS-7/home:oojah:mqtt.repo -O "/etc/yum.repos.d/Mosquitto.repo"
[root@localhost tmp]# yum install mosquitto mosquitto-clients
[root@localhost tmp]# ls -litr /etc/yum.repos.d/Mosquitto.repo

[root@localhost tmp]# less /etc/init.d/mosquitto
#! /bin/sh
~~~~~
### BEGIN INIT INFO
# Provides: mosquitto
# Required-Start: $network $remote_fs
# Required-Stop: $network $remote_fs
# Default-Start: 3 5
# Default-Stop: 0 1 2 6
# Short-Description: Mosquitto MQTT broker
# Description: Mosquitto MQTT broker
### END INIT INFO

● Auto start setting

○ Register with daemon

[root@localhost tmp]# /sbin/chkconfig --add mosquitto
[root@localhost tmp]# /sbin/chkconfig --list mosquitto
mosquitto       0:off   1:off   2:off   3:on    4:off   5:on    6:off

○ Set to start automatically at the current runlevel

[root@localhost tmp]# /sbin/chkconfig mosquitto on
[root@localhost tmp]# /sbin/chkconfig --list mosquitto
mosquitto       0:off   1:off   2:on    3:on    4:on    5:on    6:off

● Start

[root@localhost tmp]# /etc/init.d/mosquitto start
Starting mosquitto (via systemctl):                        [  OK  ]

● Monitoring by monit

Install monit (monitoring software) and set it to start automatically when mosquitto goes down. https://easyengine.io/tutorials/monitoring/monit/

[root@localhost tmp]# cd ~
[root@localhost tmp]# wget http://mmonit.com/monit/dist/binary/5.14/monit-5.14-linux-x64.tar.gz
[root@localhost tmp]# tar zxvf monit-5.14-linux-x64.tar.gz
[root@localhost tmp]# cd monit-5.14/
[root@localhost tmp]# cp bin/monit /usr/bin/monit
[root@localhost tmp]# mkdir /etc/monit
[root@localhost tmp]# touch /etc/monit/monitrc
[root@localhost tmp]# chmod 0700 /etc/monit/monitrc 
[root@localhost tmp]# ln -s /etc/monit/monitrc /etc/monitrc
[root@localhost tmp]# wget https://gist.githubusercontent.com/rahul286/9975061/raw/1aa107e62ecaaa2dacfdb61a12f13efb6f15005b/monit -P /etc/init.d/
[root@localhost tmp]# chmod u+x /etc/init.d/monit
[root@localhost tmp]# echo "START=yes" > /etc/default/monit
[root@localhost tmp]# monit -t
[root@localhost tmp]# /sbin/chkconfig  --add monit
[root@localhost tmp]# /sbin/chkconfig  monit on
[root@localhost tmp]# /sbin/chkconfig --list monit
[root@localhost tmp]# view /etc/monit.d/mosquitto.conf
check process mosquitto with pidfile /var/run/mosquitto.pid
start = "/etc/init.d/mosquitto start"
stop = "/etc/init.d/mosquitto stop"

● Log output settings

[root@localhost tmp]# sudo mkdir /var/log/mosquitto
[root@localhost tmp]# sudo chown mosquitto /var/log/mosquitto

[root@localhost tmp]# view /etc/mosquitto/mosquitto.conf
Total 0
pid_file /var/run/mosquitto.pid

persistence true
persistence_location /var/lib/mosquitto/

log_dest syslog
log_dest file /var/log/mosquitto/mosquitto.log
#log_type debug
log_type error
log_type warning
log_type notice
log_type information
#log_type none
log_type subscribe
log_type unsubscribe
#log_type websockets
#log_type all

connection_messages true

log_timestamp true

include_dir /etc/mosquitto/conf.d

[root@localhost tmp]# /etc/init.d/mosquitto reload

■ Operation check

● Publish and subscribe test

Launch two terminals and execute the following commands for each.

From the topic sensors / temperature, publish the value 32 to yourself with qos1 and subscribe.

#Subscribe
[root@localhost tmp]# mosquitto_sub -t sensors/temperature -q 

#Publish
[root@localhost tmp]# mosquitto_pub -t sensors/temperature -m 32 -q 1

#If you want to send a message that includes a line feed code-You can publish a message containing a line feed code by publishing the contents of the file using the f option.
[root@localhost tmp]# mosquitto_pub -t sensors/temperature -f /var/tmp/test.txt

http://mosquitto.org/man/mosquitto_sub-1.html http://mosquitto.org/man/mosquitto_pub-1.html

■ Operate MQTT using Python

The following will be helpful. https://librabuch.jp/2015/09/mosquiito_paho_python_mqtt/

● Introduction of pip

Install pip, which manages python packages. It's easy because you can install python using pip with one command.

[root@localhost opt]# curl -kL https://bootstrap.pypa.io/get-pip.py | python curl -kL https://bootstrap.pypa.io/get-pip.py | python
  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current
                                 Dload  Upload   Total   Spent    Left  Speed
100 1487k  100 1487k    0     0  1631k      0 --:--:-- --:--:-- --:--:-- 1630k
Collecting pip
  Downloading pip-8.1.1-py2.py3-none-any.whl (1.2MB)
    100% |████████████████████████████████| 1.2MB 333kB/s 
Collecting wheel
  Downloading wheel-0.29.0-py2.py3-none-any.whl (66kB)
    100% |████████████████████████████████| 71kB 1.9MB/s 
Installing collected packages: pip, wheel
Successfully installed pip-8.1.1 wheel-0.29.0

● Installation of paho

paho is an Eclipse library that provides MQTT functionality.

https://eclipse.org/paho/ https://pypi.python.org/pypi/paho-mqtt/1.1

[root@localhost opt]# pip install paho-mqtt
Collecting paho-mqtt
  Downloading paho-mqtt-1.1.tar.gz (41kB)
    100% |████████████████████████████████| 51kB 3.4MB/s 
Building wheels for collected packages: paho-mqtt
  Running setup.py bdist_wheel for paho-mqtt ... done
  Stored in directory: /root/.cache/pip/wheels/97/db/5f/1ddca8ee2f9b58f9bb68208323bd39bb0b177f32f434aa4b95
Successfully built paho-mqtt
Installing collected packages: paho-mqtt
Successfully installed paho-mqtt-1.1

[root@localhost opt]# ls -litr /usr/lib/python2.7/site-packages/paho/mqtt
Total 196
135479923 -rw-r--r--.1 root root 8713 March 19 23:15 publish.py
135479924 -rw-r--r--.1 root root 91388 March 19 23:15 client.py
135479925 -rw-r--r--.1 root root 20 March 19 23:15 __init__.py
135479926 -rw-r--r--.1 root root 170 March 19 23:15 __init__.pyc
135479927 -rw-r--r--.1 root root 71288 March 19 23:15 client.pyc
135479928 -rw-r--r--.1 root root 8332 March 19 23:15 publish.pyc

● Operation check

Prepare publishers and subscribers as well.


from time import sleep
import paho.mqtt.client as mqtt

HOST = '127.0.0.1'
PORT = 1883
KEEP_ALIVE = 60
TOPIC = 'test_topic/test1'
MESSAGE = 'test message'

PUBLISH_NUMBER = 5
SLEEP_TIME = 5

def publish_many_times(client, topic='topic/default', message='default', number=1, time=1, print_flag=False):

    for i in range(number):
        client.publish(topic, message)
        if print_flag == True:
            print (topic + ' ' + message)
        sleep(time)

    client.disconnect()

if __name__ == '__main__':
    client = mqtt.Client(protocol=mqtt.MQTTv311)

    print "publish start " + str(type(client))

    client.connect(HOST, port=PORT, keepalive=KEEP_ALIVE)

    publish_many_times(client,TOPIC, MESSAGE, PUBLISH_NUMBER, SLEEP_TIME)                                      
# -*- coding: utf-8 -*-
import paho.mqtt.client as mqtt

HOST = '127.0.0.1'
PORT = 1883
KEEP_ALIVE = 60
TOPIC = 'test_topic/test1'

"""
Run when trying to connect
def on_connect(client, userdata, flags, respons_code):

* client
Instance of Client class

* userdata
When creating an instance of a new Client class with any type of data>Can be set

* flags
Dictionary with response flags
Valid for users who have a clean session set to 0.
Determine if the session still exists.
If the clean session is 0, reconnect to the previously connected user.

0 :Session does not exist
1 :Session exists

* respons_code
The response code indicates whether the connection was successful.
0:Successful connection
1:Connection Failed- incorrect protocol version
2:Connection Failed- invalid client identifier
3:Connection Failed- server unavailable
4:Connection Failed- bad username or password
5:Connection Failed- not authorised
"""
def on_connect(client, userdata, flags, respons_code):
    print('status {0}'.format(respons_code))
    client.subscribe(client.topic)

"""
def on_message(client, userdata, message):
Execute when topic is received
"""
def on_message(client, userdata, message):
    print(message.topic + ' ' + str(message.payload))

if __name__ == '__main__':

    client = mqtt.Client(protocol=mqtt.MQTTv311)
    client.topic = TOPIC

    client.on_connect = on_connect
    client.on_message = on_message

    client.connect(HOST, port=PORT, keepalive=KEEP_ALIVE)

    #loop
    client.loop_forever()                                  
[root@localhost tmp]# python publisher.py 
[root@localhost tmp]# python subscriber.py 
status 0
test_topic/test1 test message
test_topic/test1 test message
test_topic/test1 test message

client.on_connect () and client.on_message () are callback functions, In the loop inside client.loop_forever (), it is called and executed by the handler.

See here for callbacks.

def _handle_connack(self):
    if self._strict_protocol:
        if self._in_packet['remaining_length'] != 2:
            return MQTT_ERR_PROTOCOL

    if len(self._in_packet['packet']) != 2:
        return MQTT_ERR_PROTOCOL

    (flags, result) = struct.unpack("!BB", self._in_packet['packet'])
    if result == CONNACK_REFUSED_PROTOCOL_VERSION and self._protocol == MQTTv311:
        self._easy_log(MQTT_LOG_DEBUG, "Received CONNACK ("+str(flags)+", "+str(result)+"), attempting downgrade to MQTT v3.1.")
        # Downgrade to MQTT v3.1
        self._protocol = MQTTv31
        return self.reconnect()

    if result == 0:
        self._state = mqtt_cs_connected

    self._easy_log(MQTT_LOG_DEBUG, "Received CONNACK ("+str(flags)+", "+str(result)+")")
    self._callback_mutex.acquire()
    if self.on_connect:
        self._in_callback = True

        if sys.version_info[0] < 3:
            argcount = self.on_connect.func_code.co_argcount
        else:
            argcount = self.on_connect.__code__.co_argcount

        if argcount == 3:
            self.on_connect(self, self._userdata, result)
        else:
            flags_dict = dict()
            flags_dict['session present'] = flags & 0x01
            self.on_connect(self, self._userdata, flags_dict, result)
        self._in_callback = False
    self._callback_mutex.release()
    if result == 0:
        rc = 0
        self._out_message_mutex.acquire()
        for m in self._out_messages:
            m.timestamp = time.time()
            if m.state == mqtt_ms_queued:
                self.loop_write() # Process outgoing messages that have just been queued up
                self._out_message_mutex.release()
                return MQTT_ERR_SUCCESS

            if m.qos == 0:
                self._in_callback = True # Don't call loop_write after _send_publish()
                rc = self._send_publish(m.mid, m.topic, m.payload, m.qos, m.retain, m.dup)
                self._in_callback = False
                if rc != 0:
                    self._out_message_mutex.release()
                    return rc
            elif m.qos == 1:
                if m.state == mqtt_ms_publish:
                    self._inflight_messages = self._inflight_messages + 1
                    m.state = mqtt_ms_wait_for_puback
                    self._in_callback = True # Don't call loop_write after _send_publish()
                    rc = self._send_publish(m.mid, m.topic, m.payload, m.qos, m.retain, m.dup)
                    self._in_callback = False
                    if rc != 0:
                        self._out_message_mutex.release()
                        return rc
            elif m.qos == 2:
                if m.state == mqtt_ms_publish:
                    self._inflight_messages = self._inflight_messages + 1
                    m.state = mqtt_ms_wait_for_pubrec
                    self._in_callback = True # Don't call loop_write after _send_publish()
                    rc = self._send_publish(m.mid, m.topic, m.payload, m.qos, m.retain, m.dup)
                    self._in_callback = False
                    if rc != 0:
                        self._out_message_mutex.release()
                        return rc
                elif m.state == mqtt_ms_resend_pubrel:
                    self._inflight_messages = self._inflight_messages + 1
                    m.state = mqtt_ms_wait_for_pubcomp
                    self._in_callback = True # Don't call loop_write after _send_pubrel()
                    rc = self._send_pubrel(m.mid, m.dup)
                    self._in_callback = False
                    if rc != 0:
                        self._out_message_mutex.release()
                        return rc
            self.loop_write() # Process outgoing messages that have just been queued up
        self._out_message_mutex.release()
        return rc
    elif result > 0 and result < 6:
        return MQTT_ERR_CONN_REFUSED
    else:
        return MQTT_ERR_PROTOCOL
def _handle_on_message(self, message):
    self._callback_mutex.acquire()
    matched = False
    for t in self.on_message_filtered:
        if topic_matches_sub(t[0], message.topic):
            self._in_callback = True
            t[1](self,self._userdata,message)
            self._in_callback = False
            matched = True

    if matched == False and self.on_message:
        self._in_callback = True
        self.on_message(self,self._userdata,message)
        self._in_callback = False

    self._callback_mutex.release()

■ Reference site

Recommended Posts

Introduction to MQTT (Introduction)
Introduction to Scrapy (1)
Introduction to Scrapy (3)
Introduction to Supervisor
Introduction to Tkinter 1: Introduction
Introduction to PyQt
Introduction to Scrapy (2)
[Linux] Introduction to Linux
Introduction to Scrapy (4)
Introduction to discord.py (2)
Introduction to discord.py
Introduction to Lightning pytorch
Introduction to Web Scraping
Introduction to Nonparametric Bayes
Introduction to EV3 / MicroPython
Introduction to Python language
Introduction to TensorFlow-Image Recognition
Introduction to OpenCV (python)-(2)
Introduction to PyQt4 Part 1
Introduction to Dependency Injection
Introduction to Private Chainer
Introduction to machine learning
AOJ Introduction to Programming Topic # 1, Topic # 2, Topic # 3, Topic # 4
Introduction to electronic paper modules
A quick introduction to pytest-mock
Introduction to dictionary lookup algorithm
Introduction to Monte Carlo Method
[Learning memorandum] Introduction to vim
Introduction to PyTorch (1) Automatic differentiation
opencv-python Introduction to image processing
Introduction to Python Django (2) Win
Introduction to Cython Writing [Notes]
An introduction to private TensorFlow
Kubernetes Scheduler Introduction to Homebrew
An introduction to machine learning
[Introduction to cx_Oracle] Overview of cx_Oracle
A super introduction to Linux
Introduction
AOJ Introduction to Programming Topic # 7, Topic # 8
[Introduction to pytorch-lightning] First Lit ♬
Introduction to Anomaly Detection 1 Basics
Introduction to RDB with sqlalchemy Ⅰ
[Introduction to Systre] Fibonacci Retracement ♬
Introduction to Nonlinear Optimization (I)
Introduction to serial communication [Python]
AOJ Introduction to Programming Topic # 5, Topic # 6
Introduction to Deep Learning ~ Learning Rules ~
[Introduction to Python] <list> [edit: 2020/02/22]
Introduction to Python (Python version APG4b)
An introduction to Python Programming
[Introduction to cx_Oracle] (8th) cx_Oracle 8.0 release
Introduction to discord.py (3) Using voice
An introduction to Bayesian optimization
Deep Reinforcement Learning 1 Introduction to Reinforcement Learning
Super introduction to machine learning
Introduction to Ansible Part ③'Inventory'
Series: Introduction to cx_Oracle Contents
[Introduction] How to use open3d
Introduction to Python For, While
Introduction to Deep Learning ~ Backpropagation ~
Introduction to Ansible Part ④'Variable'