Mercurial > prosody-modules

comparison mod_pubsub_mqtt/mqtt.lib.lua @ 5833:58df53eefa28

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
mod_pubsub_mqtt: Update to MQTT 3.1.1
author Matthew Wild <mwild1@gmail.com>
date Wed, 07 Feb 2024 11:57:30 +0000
parents d2a84e6aed2b
children ef625180a869
comparison
equal deleted inserted replaced
5832:5afc8273c5ef 5833:58df53eefa28
1 local bit = require "bit"; 1 local bit = require "util.bitcompat";
2 2
3 local stream_mt = {}; 3 local stream_mt = {};
4 stream_mt.__index = stream_mt; 4 stream_mt.__index = stream_mt;
5 5
6 function stream_mt:read_bytes(n_bytes) 6 function stream_mt:read_bytes(n_bytes)
27 local len1, len2 = self:read_bytes(2):byte(1,2); 27 local len1, len2 = self:read_bytes(2):byte(1,2);
28 local len = bit.lshift(len1, 8) + len2; 28 local len = bit.lshift(len1, 8) + len2;
29 return self:read_bytes(len), len+2; 29 return self:read_bytes(len), len+2;
30 end 30 end
31 31
32 function stream_mt:read_word()
33 local len1, len2 = self:read_bytes(2):byte(1,2);
34 local result = bit.lshift(len1, 8) + len2;
35 module:log("debug", "read_word(%02x, %02x) = %04x (%d)", len1, len2, result, result);
36 return result;
37 end
38
39 local function hasbit(byte, n_bit)
40 return bit.band(byte, 2^n_bit) ~= 0;
41 end
42
43 local function encode_string(str)
44 return string.char(bit.band(#str, 0xff00), bit.band(#str, 0x00ff))..str;
45 end
46
32 local packet_type_codes = { 47 local packet_type_codes = {
33 "connect", "connack", 48 "connect", "connack",
34 "publish", "puback", "pubrec", "pubrel", "pubcomp", 49 "publish", "puback", "pubrec", "pubrel", "pubcomp",
35 "subscribe", "subak", "unsubscribe", "unsuback", 50 "subscribe", "suback", "unsubscribe", "unsuback",
36 "pingreq", "pingresp", 51 "pingreq", "pingresp",
37 "disconnect" 52 "disconnect"
38 }; 53 };
39 54
40 function stream_mt:read_packet() 55 function stream_mt:read_packet()
57 if self:read_string() ~= "MQTT" then 72 if self:read_string() ~= "MQTT" then
58 module:log("warn", "Unexpected packet signature!"); 73 module:log("warn", "Unexpected packet signature!");
59 packet.type = nil; -- Invalid packet 74 packet.type = nil; -- Invalid packet
60 else 75 else
61 packet.version = self:read_bytes(1):byte(); 76 packet.version = self:read_bytes(1):byte();
62 packet.connect_flags = self:read_bytes(1):byte(); 77 module:log("debug", "ver: %02x", packet.version);
63 packet.keepalive_timer = self:read_bytes(1):byte(); 78 if packet.version ~= 0x04 then
79 module:log("warn", "MQTT version mismatch (got %02x, we support %02x", packet.version, 0x04);
80 end
81 local flags = self:read_bytes(1):byte();
82 module:log("debug", "flags: %02x", flags);
83 packet.keepalive_timer = self:read_bytes(2):byte();
84 module:log("debug", "keepalive: %d", packet.keepalive_timer);
85 packet.connect_flags = {};
64 length = length - 11; 86 length = length - 11;
87 packet.connect_flags = {
88 clean_session = hasbit(flags, 1);
89 will = hasbit(flags, 2);
90 will_qos = bit.band(bit.rshift(flags, 2), 0x02);
91 will_retain = hasbit(flags, 5);
92 user_name = hasbit(flags, 7);
93 password = hasbit(flags, 6);
94 };
95 module:log("debug", "%s", require "util.serialization".serialize(packet.connect_flags, "debug"));
96 module:log("debug", "Reading client_id...");
97 packet.client_id = self:read_string();
98 if packet.connect_flags.will then
99 module:log("debug", "Reading will...");
100 packet.will = {
101 topic = self:read_string();
102 message = self:read_string();
103 qos = packet.connect_flags.will_qos;
104 retain = packet.connect_flags.will_retain;
105 };
106 end
107 if packet.connect_flags.user_name then
108 module:log("debug", "Reading username...");
109 packet.username = self:read_string();
110 end
111 if packet.connect_flags.password then
112 module:log("debug", "Reading password...");
113 packet.password = self:read_string();
114 end
115 module:log("debug", "Done parsing connect!");
116 length = 0; -- No payload left
65 end 117 end
66 elseif packet.type == "publish" then 118 elseif packet.type == "publish" then
67 packet.topic = self:read_string(); 119 packet.topic = self:read_string();
68 length = length - (#packet.topic+2); 120 length = length - (#packet.topic+2);
69 if packet.qos == 1 or packet.qos == 2 then 121 if packet.qos == 1 or packet.qos == 2 then
85 packet.topics = topics; 137 packet.topics = topics;
86 end 138 end
87 if length > 0 then 139 if length > 0 then
88 packet.data = self:read_bytes(length); 140 packet.data = self:read_bytes(length);
89 end 141 end
142 module:log("debug", "MQTT packet complete!");
90 return packet; 143 return packet;
91 end 144 end
92 145
93 local function new_parser(self) 146 local function new_parser(self)
94 return coroutine.wrap(function (data) 147 return coroutine.wrap(function (data)
100 end 153 end
101 end); 154 end);
102 end 155 end
103 156
104 function stream_mt:feed(data) 157 function stream_mt:feed(data)
105 module:log("debug", "Feeding %d bytes", #data);
106 local packets = {}; 158 local packets = {};
107 local packet = self.parser(data); 159 local packet = self.parser(data);
108 while packet do 160 while packet do
109 module:log("debug", "Received packet"); 161 module:log("debug", "Received packet");
110 table.insert(packets, packet); 162 table.insert(packets, packet);
133 if packet.type == "publish" then 185 if packet.type == "publish" then
134 local topic = packet.topic or ""; 186 local topic = packet.topic or "";
135 packet.data = string.char(bit.band(#topic, 0xff00), bit.band(#topic, 0x00ff))..topic..packet.data; 187 packet.data = string.char(bit.band(#topic, 0xff00), bit.band(#topic, 0x00ff))..topic..packet.data;
136 elseif packet.type == "suback" then 188 elseif packet.type == "suback" then
137 local t = {}; 189 local t = {};
138 for _, topic in ipairs(packet.topics) do 190 for i, result_code in ipairs(packet.results) do
139 table.insert(t, string.char(bit.band(#topic, 0xff00), bit.band(#topic, 0x00ff))..topic.."\000"); 191 table.insert(t, string.char(result_code));
140 end 192 end
141 packet.data = table.concat(t); 193 packet.data = packet.id..table.concat(t);
142 end 194 end
143 195
144 -- Get length 196 -- Get length
145 local length = #(packet.data or ""); 197 local length = #(packet.data or "");
146 repeat 198 repeat