From 8c3006701110a2d9377b6d17d20805b10d6d6f57 Mon Sep 17 00:00:00 2001 From: Pablo Ventura Date: Thu, 16 Jul 2026 13:02:22 -0300 Subject: [PATCH 1/5] umqtt.simple: Encode subscribe Remaining Length as VBI. Use the same variable-byte Remaining Length encoding as publish for subscribe and unsubscribe so long topics no longer overflow a single length byte. Fixes #969. Signed-off-by: Pablo Ventura --- micropython/umqtt.simple/test_umqtt_simple.py | 76 +++++++++++++++++++ micropython/umqtt.simple/umqtt/simple.py | 32 ++++++-- 2 files changed, 102 insertions(+), 6 deletions(-) create mode 100644 micropython/umqtt.simple/test_umqtt_simple.py diff --git a/micropython/umqtt.simple/test_umqtt_simple.py b/micropython/umqtt.simple/test_umqtt_simple.py new file mode 100644 index 000000000..c31efd17a --- /dev/null +++ b/micropython/umqtt.simple/test_umqtt_simple.py @@ -0,0 +1,76 @@ +import io +import sys + + +class Socket: + def __init__(self, read_data=b""): + self._write_buffer = io.BytesIO() + self._read_buffer = io.BytesIO(read_data) + + def write(self, buf, length=None): + if length is None: + length = len(buf) + self._write_buffer.write(buf[:length]) + + def read(self, n): + return self._read_buffer.read(n) + + def setblocking(self, blocking): + pass + + def close(self): + pass + + +sys.path.insert(0, "micropython/umqtt.simple") +# ruff: noqa: E402 +from umqtt.simple import MQTTClient + + +def make_client(read_data): + c = MQTTClient(b"cid", "127.0.0.1") + c.sock = Socket(read_data) + c.set_callback(lambda topic, msg: None) + return c + + +def test_subscribe_short_topic(): + # Remaining Length = 5 + 4 = 9 -> single VBI byte 0x09, pid=1 + c = make_client(b"\x90\x03\x00\x01\x00") + c.subscribe(b"abcd", qos=0) + out = c.sock._write_buffer.getvalue() + assert out[:2] == b"\x82\x09", out + assert out[2:4] == b"\x00\x01", out + assert out[4:6] == b"\x00\x04", out + assert out[6:10] == b"abcd", out + assert out[10:11] == b"\x00", out + + +def test_subscribe_long_topic(): + # Remaining Length = 5 + 123 = 128 -> VBI 0x80 0x01 + topic = b"a" * 123 + c = make_client(b"\x90\x03\x00\x01\x00") + c.subscribe(topic, qos=0) + out = c.sock._write_buffer.getvalue() + assert out[:3] == b"\x82\x80\x01", out + assert out[3:5] == b"\x00\x01", out + assert out[5:7] == b"\x00\x7b", out + assert out[7:130] == topic, out + assert out[130:131] == b"\x00", out + + +def test_unsubscribe_long_topic(): + # Remaining Length = 4 + 123 = 127 -> still one VBI byte; use 124 for 128 + topic = b"a" * 124 + c = make_client(b"\xb0\x02\x00\x01") + c.unsubscribe(topic) + out = c.sock._write_buffer.getvalue() + assert out[:3] == b"\xa2\x80\x01", out + assert out[3:5] == b"\x00\x01", out + assert out[5:7] == b"\x00\x7c", out + assert out[7:131] == topic, out + + +test_subscribe_short_topic() +test_subscribe_long_topic() +test_unsubscribe_long_topic() diff --git a/micropython/umqtt.simple/umqtt/simple.py b/micropython/umqtt.simple/umqtt/simple.py index bfdaf503c..68cf807a0 100644 --- a/micropython/umqtt.simple/umqtt/simple.py +++ b/micropython/umqtt.simple/umqtt/simple.py @@ -162,9 +162,19 @@ def subscribe(self, topic, qos=0): assert self.cb is not None, "Subscribe callback is not set" pkt = bytearray(b"\x82\0\0\0") self.pid += 1 - struct.pack_into("!BH", pkt, 1, 2 + 2 + len(topic) + 1, self.pid) + pid = self.pid + sz = 2 + 2 + len(topic) + 1 + assert sz < 2097152 + i = 1 + while sz > 0x7F: + pkt[i] = (sz & 0x7F) | 0x80 + sz >>= 7 + i += 1 + pkt[i] = sz # print(hex(len(pkt)), hexlify(pkt, ":")) - self.sock.write(pkt) + self.sock.write(pkt, i + 1) + struct.pack_into("!H", pkt, 0, pid) + self.sock.write(pkt, 2) self._send_str(topic) self.sock.write(qos.to_bytes(1, "little")) while 1: @@ -172,7 +182,7 @@ def subscribe(self, topic, qos=0): if op == 0x90: resp = self.sock.read(4) # print(resp) - assert resp[1] == pkt[2] and resp[2] == pkt[3] + assert resp[1] == pid >> 8 and resp[2] == (pid & 0xFF) if resp[3] == 0x80: raise MQTTException(resp[3]) return @@ -180,14 +190,24 @@ def subscribe(self, topic, qos=0): def unsubscribe(self, topic): pkt = bytearray(b"\xa2\0\0\0") self.pid += 1 - struct.pack_into("!BH", pkt, 1, 2 + 2 + len(topic), self.pid) - self.sock.write(pkt) + pid = self.pid + sz = 2 + 2 + len(topic) + assert sz < 2097152 + i = 1 + while sz > 0x7F: + pkt[i] = (sz & 0x7F) | 0x80 + sz >>= 7 + i += 1 + pkt[i] = sz + self.sock.write(pkt, i + 1) + struct.pack_into("!H", pkt, 0, pid) + self.sock.write(pkt, 2) self._send_str(topic) while 1: op = self.wait_msg() if op == 0xB0: resp = self.sock.read(3) - assert resp[1] == pkt[2] and resp[2] == pkt[3] + assert resp[1] == pid >> 8 and resp[2] == (pid & 0xFF) return # Wait for a single incoming MQTT message and process it. From e178efef2d462da92f96cf040b007d23ded66b1f Mon Sep 17 00:00:00 2001 From: Pablo Ventura Date: Mon, 20 Jul 2026 10:49:39 -0300 Subject: [PATCH 2/5] umqtt.simple: Share subscribe/unsubscribe packet encoding. Factor Remaining Length VBI and ACK wait into _send_subunsub to reduce .mpy size after the long-topic fix. Signed-off-by: Pablo Ventura --- micropython/umqtt.simple/umqtt/simple.py | 45 ++++++++---------------- 1 file changed, 14 insertions(+), 31 deletions(-) diff --git a/micropython/umqtt.simple/umqtt/simple.py b/micropython/umqtt.simple/umqtt/simple.py index 68cf807a0..e7967f41e 100644 --- a/micropython/umqtt.simple/umqtt/simple.py +++ b/micropython/umqtt.simple/umqtt/simple.py @@ -158,12 +158,12 @@ def publish(self, topic, msg, retain=False, qos=0): elif qos == 2: assert 0 - def subscribe(self, topic, qos=0): - assert self.cb is not None, "Subscribe callback is not set" - pkt = bytearray(b"\x82\0\0\0") + def _send_subunsub(self, topic, typ, ack_op, ack_n, qos=None): + pkt = bytearray(4) + pkt[0] = typ self.pid += 1 pid = self.pid - sz = 2 + 2 + len(topic) + 1 + sz = 4 + len(topic) + (0 if qos is None else 1) assert sz < 2097152 i = 1 while sz > 0x7F: @@ -171,44 +171,27 @@ def subscribe(self, topic, qos=0): sz >>= 7 i += 1 pkt[i] = sz - # print(hex(len(pkt)), hexlify(pkt, ":")) self.sock.write(pkt, i + 1) struct.pack_into("!H", pkt, 0, pid) self.sock.write(pkt, 2) self._send_str(topic) - self.sock.write(qos.to_bytes(1, "little")) + if qos is not None: + self.sock.write(qos.to_bytes(1, "little")) while 1: op = self.wait_msg() - if op == 0x90: - resp = self.sock.read(4) - # print(resp) + if op == ack_op: + resp = self.sock.read(ack_n) assert resp[1] == pid >> 8 and resp[2] == (pid & 0xFF) - if resp[3] == 0x80: + if ack_n == 4 and resp[3] == 0x80: raise MQTTException(resp[3]) return + def subscribe(self, topic, qos=0): + assert self.cb is not None, "Subscribe callback is not set" + self._send_subunsub(topic, 0x82, 0x90, 4, qos) + def unsubscribe(self, topic): - pkt = bytearray(b"\xa2\0\0\0") - self.pid += 1 - pid = self.pid - sz = 2 + 2 + len(topic) - assert sz < 2097152 - i = 1 - while sz > 0x7F: - pkt[i] = (sz & 0x7F) | 0x80 - sz >>= 7 - i += 1 - pkt[i] = sz - self.sock.write(pkt, i + 1) - struct.pack_into("!H", pkt, 0, pid) - self.sock.write(pkt, 2) - self._send_str(topic) - while 1: - op = self.wait_msg() - if op == 0xB0: - resp = self.sock.read(3) - assert resp[1] == pid >> 8 and resp[2] == (pid & 0xFF) - return + self._send_subunsub(topic, 0xA2, 0xB0, 3) # Wait for a single incoming MQTT message and process it. # Subscribed messages are delivered to a callback previously From f20a86ad25d3f47d897c0b3327c6030e3e18f95f Mon Sep 17 00:00:00 2001 From: Pablo Ventura Date: Mon, 20 Jul 2026 10:55:00 -0300 Subject: [PATCH 3/5] umqtt.simple: Share Remaining Length VBI encoding. Factor the Variable Byte Integer encoder into a module helper and reuse it from connect, publish, and the subscribe/unsubscribe path to cut .mpy size. Signed-off-by: Pablo Ventura --- micropython/umqtt.simple/umqtt/simple.py | 41 ++++++++++-------------- 1 file changed, 17 insertions(+), 24 deletions(-) diff --git a/micropython/umqtt.simple/umqtt/simple.py b/micropython/umqtt.simple/umqtt/simple.py index e7967f41e..a8016f376 100644 --- a/micropython/umqtt.simple/umqtt/simple.py +++ b/micropython/umqtt.simple/umqtt/simple.py @@ -7,6 +7,16 @@ class MQTTException(Exception): pass +def _encode_len(pkt, sz): + i = 1 + while sz > 0x7F: + pkt[i] = (sz & 0x7F) | 0x80 + sz >>= 7 + i += 1 + pkt[i] = sz + return i + + class MQTTClient: def __init__( self, @@ -93,12 +103,7 @@ def connect(self, clean_session=True, timeout=None): msg[6] |= 0x4 | (self.lw_qos & 0x1) << 3 | (self.lw_qos & 0x2) << 3 msg[6] |= self.lw_retain << 5 - i = 1 - while sz > 0x7F: - premsg[i] = (sz & 0x7F) | 0x80 - sz >>= 7 - i += 1 - premsg[i] = sz + i = _encode_len(premsg, sz) self.sock.write(premsg, i + 2) self.sock.write(msg) @@ -130,12 +135,7 @@ def publish(self, topic, msg, retain=False, qos=0): if qos > 0: sz += 2 assert sz < 2097152 - i = 1 - while sz > 0x7F: - pkt[i] = (sz & 0x7F) | 0x80 - sz >>= 7 - i += 1 - pkt[i] = sz + i = _encode_len(pkt, sz) # print(hex(len(pkt)), hexlify(pkt, ":")) self.sock.write(pkt, i + 1) self._send_str(topic) @@ -163,26 +163,19 @@ def _send_subunsub(self, topic, typ, ack_op, ack_n, qos=None): pkt[0] = typ self.pid += 1 pid = self.pid - sz = 4 + len(topic) + (0 if qos is None else 1) - assert sz < 2097152 - i = 1 - while sz > 0x7F: - pkt[i] = (sz & 0x7F) | 0x80 - sz >>= 7 - i += 1 - pkt[i] = sz + i = _encode_len(pkt, 4 + len(topic) + (qos != None)) self.sock.write(pkt, i + 1) struct.pack_into("!H", pkt, 0, pid) self.sock.write(pkt, 2) self._send_str(topic) - if qos is not None: - self.sock.write(qos.to_bytes(1, "little")) + if qos != None: + self.sock.write(bytes((qos,))) while 1: op = self.wait_msg() if op == ack_op: resp = self.sock.read(ack_n) - assert resp[1] == pid >> 8 and resp[2] == (pid & 0xFF) - if ack_n == 4 and resp[3] == 0x80: + assert (resp[1] << 8 | resp[2]) == pid + if ack_n > 3 and resp[3] == 0x80: raise MQTTException(resp[3]) return From 5e49b1bd41d312d9d2a8e4f19d7bd6a918896abc Mon Sep 17 00:00:00 2001 From: Pablo Ventura Date: Mon, 20 Jul 2026 11:06:03 -0300 Subject: [PATCH 4/5] umqtt.simple: Avoid None comparisons in _send_subunsub. Use ack_n to decide whether to include the QoS byte so ruff E711 passes without growing the compiled .mpy size. Signed-off-by: Pablo Ventura --- micropython/umqtt.simple/manifest.py | 2 +- micropython/umqtt.simple/umqtt/simple.py | 6 +++--- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/micropython/umqtt.simple/manifest.py b/micropython/umqtt.simple/manifest.py index efbda68f5..2e4d33abb 100644 --- a/micropython/umqtt.simple/manifest.py +++ b/micropython/umqtt.simple/manifest.py @@ -1,4 +1,4 @@ -metadata(description="Lightweight MQTT client for MicroPython.", version="1.7.1") +metadata(description="Lightweight MQTT client for MicroPython.", version="1.8.0") # Originally written by Paul Sokolovsky. diff --git a/micropython/umqtt.simple/umqtt/simple.py b/micropython/umqtt.simple/umqtt/simple.py index a8016f376..7539711e3 100644 --- a/micropython/umqtt.simple/umqtt/simple.py +++ b/micropython/umqtt.simple/umqtt/simple.py @@ -158,17 +158,17 @@ def publish(self, topic, msg, retain=False, qos=0): elif qos == 2: assert 0 - def _send_subunsub(self, topic, typ, ack_op, ack_n, qos=None): + def _send_subunsub(self, topic, typ, ack_op, ack_n, qos=0): pkt = bytearray(4) pkt[0] = typ self.pid += 1 pid = self.pid - i = _encode_len(pkt, 4 + len(topic) + (qos != None)) + i = _encode_len(pkt, 4 + len(topic) + (ack_n > 3)) self.sock.write(pkt, i + 1) struct.pack_into("!H", pkt, 0, pid) self.sock.write(pkt, 2) self._send_str(topic) - if qos != None: + if ack_n > 3: self.sock.write(bytes((qos,))) while 1: op = self.wait_msg() From 6a9f163da4b80fa20853a5ab7acfb60cb884900e Mon Sep 17 00:00:00 2001 From: Pablo Ventura Date: Mon, 27 Jul 2026 10:49:00 -0300 Subject: [PATCH 5/5] tools/ci.sh: Add umqtt.simple mock test to package tests CI. Signed-off-by: Pablo Ventura --- tools/ci.sh | 1 + 1 file changed, 1 insertion(+) diff --git a/tools/ci.sh b/tools/ci.sh index 2fc7ba004..4c56fb9c5 100755 --- a/tools/ci.sh +++ b/tools/ci.sh @@ -57,6 +57,7 @@ function ci_package_tests_run { export MICROPYPATH for test in \ micropython/drivers/storage/sdcard/sdtest.py \ + micropython/umqtt.simple/test_umqtt_simple.py \ micropython/xmltok/test_xmltok.py \ python-ecosys/requests/test_requests.py \ python-stdlib/argparse/test_argparse.py \