From 635d100dedb212a49cfc753b8571cdec85f039f2 Mon Sep 17 00:00:00 2001 From: Pablo Ventura Date: Thu, 16 Jul 2026 13:02:22 -0300 Subject: [PATCH 1/4] 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 51bbbaa67dae280e4cf90739b26d6f7155da806f Mon Sep 17 00:00:00 2001 From: Pablo Ventura Date: Mon, 20 Jul 2026 10:49:39 -0300 Subject: [PATCH 2/4] 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 2237dc9db4ed113a354e20bcfa8c3a75fa51ea74 Mon Sep 17 00:00:00 2001 From: Pablo Ventura Date: Mon, 20 Jul 2026 10:55:00 -0300 Subject: [PATCH 3/4] 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 4309a60ed82adc42c9f661cb4f4f5393192cc5fd Mon Sep 17 00:00:00 2001 From: Pablo Ventura Date: Mon, 20 Jul 2026 11:06:03 -0300 Subject: [PATCH 4/4] 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/umqtt/simple.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) 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()