بروتوكول 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

حين تكون القيمة واحدة يكفي نص قصير. وحين تزيد، نكتب الحمولة بصيغة 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.

أوامر الفصل:

قاموس المصطلحات – מילון מונחים:

أسئلة للتمرين: