פרוטוקול MQTT

16.1 מתווך במקום חיבור ישיר (Publish and Subscribe)

ב-HTTP מי שפונה מכיר את הכתובת של מי שהוא פונה אליו, והצד השני צריך להיות מוכן באותו רגע. MQTT מפריד בין שני הצדדים: הלוח שמודד אינו יודע מי קורא, ומי שקורא אינו יודע מי מדד.

ההודעה ב-MQTT היא רצף בייטים שהפרוטוקול אינו קובע את משמעותו, ואנחנו קוראים לה מטען. כל התקן מתחבר למתווך דרך TCP והחיבור נשאר פתוח, ולכן המתווך יכול להעביר פקודה ללוח בכל רגע, בלי שהלוח ישאל עליה.

16.2 שמות נושאים ותווים כלליים (Topics and Wildcards)

הנושא הוא נתיב שהקווים האלכסוניים מפרידים בין דרגותיו, ואנחנו בוחרים את המבנה שלו. הכלל שלנו בספר הזה: מהכללי אל הפרטי, כלומר school/desk1/pot, באותיות אנגליות קטנות, בלי רווחים ובלי קו אלכסוני בתחילתו. הפקודה והמצב הם שני נושאים נפרדים: כותבים .../led/set לפקודה ו-.../led/state למצב בפועל.

רק בהרשמה אפשר להשתמש בשני תווים כלליים:

import time
from umqtt.simple import MQTTClient
import wifi

TOPICS = (b"school/lab1/temp", b"school/lab2/temp",
          b"school/lab1/hum", b"school/lab1/door/temp")


def on_message(topic, msg):
    print("  received:", topic.decode(), "=", msg.decode())


wifi.connect()
client = MQTTClient("desk1-wild", "broker.hivemq.com")
client.set_callback(on_message)
client.connect()
client.subscribe(b"school/+/temp")

for t in TOPICS:
    print("publish:", t.decode())
    client.publish(t, b"21")
    time.sleep(1)
    client.check_msg()

client.disconnect()

publish: school/lab1/temp
  received: school/lab1/temp = 21
publish: school/lab2/temp
  received: school/lab2/temp = 21
publish: school/lab1/hum
publish: school/lab1/door/temp

התוכנית מדברת עם עצמה דרך המתווך, וזו דרך טובה לבדוק כל הרשמה. שימו לב ל-check_msg() אחרי כל פרסום: ההודעה מגיעה לפונקציה רק אם מבקשים אותה. נחזור לזה בסעיף 16.4.

16.3 פרסום קריאה (Publishing)

נשתמש בספרייה umqtt.simple מתוך micropython-lib. קובצי ההגדרה הרשמיים של ESP32 מזכירים אותה בתוך הקושחה, וקובצי Pico W אינם מזכירים אותה. לכן קודם ננסה from umqtt.simple import MQTTClient. אם מופיעה שגיאה שהמודול אינו קיים, נתחבר לרשת ונתקין אותו פעם אחת בפונקציה mip.install('umqtt.simple'), כפי שראינו בפרק 11.

כל התוכניות בפרק הזה צריכות קודם כול רשת, ולשם כך אנחנו משתמשים בספרייה wifi.py מפרק 14.

import time
from umqtt.simple import MQTTClient
import board
import wifi
from convert import to_percent

BROKER = "broker.hivemq.com"
TOPIC = b"school/desk1/pot"

pot = board.analog_in(board.POT)

wifi.connect()
client = MQTTClient("desk1-pub", BROKER)
client.connect()
print("connected to", BROKER)

while True:
    percent = to_percent(pot.read_u16())
    client.publish(TOPIC, str(percent).encode())
    print("sent", percent)
    time.sleep(5)

connected to broker.hivemq.com
sent 25
sent 50
sent 74
[mqtt_pub]
school/desk1/pot 25
school/desk1/pot 50
school/desk1/pot 74
...

16.4 הרשמה וקבלת פקודה (Subscribing)

הקליטה עוברת שלושה שלבים, והסדר שלהם חשוב: נותנים לספרייה פונקציה שתיקרא כשמגיעה הודעה, set_callback, אחר כך מתחברים, ואחר כך נרשמים. אחרי זה צריך לתת לספרייה הזדמנות לקרוא את מה שהגיע, באחת משתי פונקציות:

import time
from machine import Pin
from umqtt.simple import MQTTClient
import board
import wifi

CMD = b"school/desk1/led/set"
STATE = b"school/desk1/led/state"

led = Pin(board.LED, Pin.OUT)
client = MQTTClient("desk1-led", "broker.hivemq.com")


def on_message(topic, msg):
    if msg == b"on":
        led.on()
    elif msg == b"off":
        led.off()
    else:
        print("unknown command:", msg)
        return
    client.publish(STATE, msg)
    print("led is now", led.value())


wifi.connect()
client.set_callback(on_message)
client.connect()
client.subscribe(CMD)

while True:
    client.check_msg()
    time.sleep(0.1)

led is now 1
unknown command: b'blink'
led is now 0
[mqtt_pub]
school/desk1/led/state on
school/desk1/led/state off
...

16.5 הודעת JSON

כשיש ערך אחד, מספיק טקסט קצר. כשיש יותר, כותבים את המטען בפורמט JSON שראינו בפרק 12: הטלפון והשרת קוראים אותו בקלות, והוא מראה את משמעותו של כל ערך. כל התקן יכול לשלוח לנו כל דבר, ולכן מגינים על הקריאה בעזרת try.

import json
import time
from umqtt.simple import MQTTClient
import board
import wifi
from convert import to_percent

CMD = b"school/desk1/cmd"
DATA = b"school/desk1/data"

pot = board.analog_in(board.POT)
period = 4


def on_message(topic, msg):
    global period
    try:
        period = json.loads(msg)["period"]
        print("new period:", period)
    except (ValueError, KeyError):
        print("ignored:", msg)


wifi.connect()
client = MQTTClient("desk1-json", "broker.hivemq.com")
client.set_callback(on_message)
client.connect()
client.subscribe(CMD)

elapsed = 0
while True:
    client.check_msg()
    if elapsed >= period:
        elapsed = 0
        data = {"pot": to_percent(pot.read_u16())}
        client.publish(DATA, json.dumps(data).encode())
    time.sleep(1)
    elapsed += 1

new period: 2
ignored: b'hello'
[mqtt_pub]
school/desk1/data {"pot": 50}
school/desk1/data {"pot": 50}
school/desk1/data {"pot": 50}
school/desk1/data {"pot": 50}
...

16.6 QoS, הודעה שמורה, keepalive וצוואה (QoS, Retain, Keepalive, Last Will)

איכות שירות, ושמה QoS, היא מה שאנחנו דורשים ממסירת ההודעה:

קריאה שחוזרת כל 5 שניות מסתפקת ב-0: הקריאה הבאה תפצה. פקודה שלא באה אחריה פקודה אחרת ראויה ל-1, ואז התוכנית צריכה לעמוד בכך שהיא עלולה להגיע פעמיים. עוד שלושה רעיונות משלימים את התמונה:

• הודעה שמורה: עם retain=True המתווך שומר את ההודעה האחרונה בנושא, ומוסר אותה מיד לכל מי שנרשם אחר כך.
• keepalive: מתחייבים לשלוח משהו בכל פרק זמן, ואם הלוח שותק זמן רב המתווך רואה אותו כמנותק. umqtt.simple אינה שולחת את הדופק מעצמה: צריך לקרוא ל-ping() או לפרסם.
• צוואה אחרונה: הודעה שאנחנו משאירים אצל המתווך מראש, והוא מפרסם אותה אם הלוח התנתק בלי להיפרד. כך האחרים יודעים שהלוח מת, ולא רק שהוא שותק.

import time
from umqtt.simple import MQTTClient
import wifi

STATUS = b"school/desk1/status"

wifi.connect()
client = MQTTClient("desk1", "broker.hivemq.com", keepalive=60)
client.set_last_will(STATUS, b"offline", retain=True, qos=1)
client.connect()
client.publish(STATUS, b"online", retain=True, qos=1)

while True:
    client.ping()
    client.check_msg()
    time.sleep(30)

השורה החשובה היא set_last_will, והיא באה לפני connect() לפי תיעוד הספרייה. בעת ההתחברות מפרסמים online כהודעה שמורה באותו נושא, וכך מי שנרשם בכל זמן יודע את המצב עכשיו: online או offline.

16.7 התאוששות מניתוק (Reconnecting)

הלוח יעבוד ימים, והרשת תתנתק. בלי טיפול, התוכנית נעצרת בשגיאת OSError הראשונה ואינה חוזרת. הרעיון פשוט: תופסים את השגיאה, מנסים שוב, וממתינים בין הניסיונות זמן שמוכפל, כדי לא להציף את המתווך כשהוא חוזר.

import time
from umqtt.simple import MQTTClient
import wifi

TOPIC = b"school/desk1/count"
client = MQTTClient("desk1-safe", "broker.hivemq.com")


def reconnect(tries=4):
    wait = 1
    for _ in range(tries):
        try:
            client.connect()
            print("connected")
            return True
        except OSError:
            print("no broker, waiting", wait, "s")
            time.sleep(wait)
            wait *= 2
    return False


def send(msg):
    try:
        client.publish(TOPIC, msg)
    except OSError:
        print("link lost")
        if reconnect():
            client.publish(TOPIC, msg)


wifi.connect()
if reconnect():
    for i in range(4):
        if i == 2:
            client.disconnect()
        send(str(i).encode())
        time.sleep(1)
else:
    print("giving up")

connected
link lost
connected

16.8 שני בקרים דרך מתווך (Two Boards)

בפרויקט הזוגי בקר אחד צריך לקבל פקודה מבקר אחר. עם מתווך, אף לוח אינו צריך את הכתובת של האחר: מספיק להסכים על שני נושאים. לוח A מפרסם פקודה כשלוחצים על הלחצן, ולוח B מבצע אותה ומפרסם את המצב. 192.168.1.10 היא כתובת לדוגמה של המתווך ברשת המקומית.

import time
from machine import Pin
from umqtt.simple import MQTTClient
import board
import wifi

BROKER = "192.168.1.10"
CMD = b"school/pair1/led/set"
STATE = b"school/pair1/led/state"

btn = Pin(board.BTN, Pin.IN, Pin.PULL_UP)
lit = False


def on_message(topic, msg):
    print("B reports:", msg.decode())


wifi.connect()
client = MQTTClient("pair1-a", BROKER)
client.set_callback(on_message)
client.connect()
client.subscribe(STATE)

last = 1
while True:
    now = btn.value()
    if last == 1 and now == 0:
        lit = not lit
        client.publish(CMD, b"on" if lit else b"off")
        print("pressed, asking for", lit)
    last = now
    client.check_msg()
    time.sleep(0.05)

pressed, asking for True
B reports: on
pressed, asking for False
B reports: off
[mqtt_pub]
school/pair1/led/set on
school/pair1/led/set off
...

from machine import Pin
from umqtt.simple import MQTTClient
import board
import wifi

BROKER = "192.168.1.10"
CMD = b"school/pair1/led/set"
STATE = b"school/pair1/led/state"

led = Pin(board.LED, Pin.OUT)
client = MQTTClient("pair1-b", BROKER)


def on_message(topic, msg):
    led.value(1 if msg == b"on" else 0)
    client.publish(STATE, msg)
    print("led:", led.value())


wifi.connect()
client.set_callback(on_message)
client.connect()
client.subscribe(CMD)

while True:
    client.wait_msg()

led: 1
led: 0
[mqtt_pub]
school/pair1/led/state on
school/pair1/led/state off
...

בבנייה כל לוח רץ בסימולציה משלו, ומה שמגיע אליו מהצד השני כתוב מראש בסימולציה. על שולחן העבודה מריצים את שתי התוכניות על שני לוחות אמיתיים, וכל אחד מהם מגיע אל האחר דרך המתווך, בלי לשנות אף שורה.

16.9 היכן רץ המתווך (Where the Broker Runs)

לניסוי הראשון מספיק מתווך ציבורי באינטרנט, כמו broker.hivemq.com, שבו השתמשנו כאן, והודעה עברה דרכו הלוך ושוב מלוח אמיתי. מתווכים ציבוריים עלולים להפסיק לעבוד בלי התראה, ולכן אם הוא לא עונה מנסים אחר, כמו broker.emqx.io. הוא לניסוי בלבד, מכמה סיבות ברורות:

• כל מי שמכיר את שם הנושא קורא את ההודעות שלנו וכותב אליו. כך כל אחד יכול לכבות את הנורית שלנו או להפעיל את המנוע שלנו.
• אין סיסמה ואין הצפנה בשער 1883, וכל נתון שאנחנו שולחים חשוף.
• איש אינו מבטיח לנו שהוא יפעל מחר, וכשהאינטרנט מתנתק כל המערכת נעצרת, גם אם שני הלוחות באותו חדר.

בפרויקט אמיתי מפעילים מתווך משלנו על Raspberry Pi 4/5 באותה רשת, בתוכנה Mosquitto. הפקודות הבאות לא הורצו בבניית הספר. אחרי ההתקנה אומרים למתווך להאזין לרשת: מגרסה 2.0 הוא מקבל כברירת מחדל רק חיבורים מההתקן עצמו.

sudo apt update
sudo apt install mosquitto mosquitto-clients

# /etc/mosquitto/conf.d/lab.conf
listener 1883
allow_anonymous true

sudo systemctl restart mosquitto
mosquitto_sub -h localhost -t 'school/#' -v
mosquitto_pub -h localhost -t school/pair1/led/set -m on

סיסמאות, הפעלה רציפה של המתווך וחיבורו למסד נתונים: זה נושא פרק 23.

עיקרי הפרק:

• כל התקן מדבר רק עם המתווך, והנושא הוא ההסכם היחיד בין המפרסם למנוי.
• מארגנים את הנושא מהכללי אל הפרטי, והפקודה והמצב הם שני נושאים נפרדים.
• הקליטה אינה מתרחשת מעצמה: קוראים ל-check_msg() או ל-wait_msg(), והמטען מגיע מסוג bytes.
• לא סומכים על שום הודעה שמגיעה: try סביב קריאת JSON.
• umqtt.simple תומכת רק באיכות שירות 0 ו-1. הצוואה האחרונה וההודעה השמורה מעדכנות את מי שמצטרף באיחור.
• ניתוק הוא חלק מהעבודה: מתחברים מחדש ונרשמים מחדש, ומכפילים את זמן ההמתנה.
• המתווך הציבורי הוא לניסוי בלבד, ובפרויקט אמיתי מתווך פרטי על Raspberry Pi 4/5.

פקודות הפרק:

מילון מונחים:

שאלות לתרגול: