feat: implement QoS 0 PUBLISH

mqtt.lua

@@ -10,32 +10,52 @@ local safeRead = function (conn, n)
 end
 
 local readVarint = function (conn, first_byte)
-    local b, err
+    local b, data, err
     if first_byte == nil then
-        b, err = safeRead(conn, 1)
+        data, err = safeRead(conn, 1)
+        if err ~= nil then
+            return nil, err
+        end
+
+        b = string.byte(data)
     else
         b = first_byte
     end
 
     local n, s = 0, 0
-    while err == nil and b & 0x80 == 0x80 do
+    while b & 0x80 == 0x80 do
         if s > 21 then
-            return 0, "number too large"
+            return nil, "number too large"
         end
 
         n = n + ((b & 0x7F) << s)
         s = s + 7
 
-        b, err = safeRead(conn, 1)
+        data, err = safeRead(conn, 1)
+        if err ~= nil then
+            return nil, err
         end
 
-    if err ~= nil then
-        return n, err
+        b = string.byte(data)
     end
 
     return n + (b << s), nil
 end
 
+local encodeVarint = function (n)
+    if n > 268435455 then
+        return nil, "number too large"
+    end
+
+    local data = ""
+    while n > 127 do
+        data = data .. string.char(0x80 | n & 0x7F)
+        n = n >> 7
+    end
+
+    return data .. string.char(n), nil
+end
+
 local MqttClient = {}
 
 function mqtt.open (address, port)
@@ -64,6 +84,10 @@ function MqttClient:new (conn)
 end
 
 function MqttClient:handle ()
+    if not (self.is_connecting or self.is_connected) then
+        return "no connection"
+    end
+
     local data, err = self.conn:safeRead(2)
     if err ~= nil then
         return err
@@ -111,12 +135,27 @@ function MqttClient:handle ()
 
         self.is_connecting = false
         self.is_connected = true
-    elseif ptype == 0x40 then -- PUBACK
+    elseif ptype & 0xF0 == 0x40 then -- PUBACK
         -- TODO
-    elseif ptype == 0xD0 then -- PINGRESP
-        -- TODO
-    elseif ptype == 0xE0 then -- DISCONNECT
+    elseif ptype & 0xF0 == 0xD0 then -- PINGRESP
         -- TODO
+    elseif ptype & 0xF0 == 0xE0 then -- DISCONNECT
+        if ptype ~= 0xE0 then
+            self:disconnect(0x81)
+            return "malformed packet"
+        end
+
+        self.is_connecting = false
+        self.is_connected = false
+
+        if length > 0 then
+            reason, _ = string.unpack("B", data)
+            if reason ~= 0 then
+                return "disconnect with error"
+            end
+        end
+
+        return "disconnect"
     end
 
     return nil
@@ -124,7 +163,7 @@ end
 
 function MqttClient:connect (username, password)
     if self.is_connecting or self.is_connected then
-        return nil
+        return "already connected"
     end
 
     local length = 13
@@ -138,11 +177,7 @@ function MqttClient:connect (username, password)
         flags = flags | 0x40
     end
 
-    if length > 127 then
-        return "packet size exceeds current implementation capabilities"
-    end
-
-    local data = string.pack("> B B s2 B B I2 B s2", 0x10, length, "MQTT", 5, flags, 0, 0, "")
+    local data = string.char(0x10) .. encodeVarint(length) .. string.pack("> s2 B B I2 B s2", "MQTT", 5, flags, 0, 0, "")
     if username ~= nil then
         data = data .. string.pack("> s2", username)
     end
@@ -165,9 +200,38 @@ function MqttClient:connect (username, password)
     return nil
 end
 
+function MqttClient:publish (topic, payload)
+    if not (self.is_connecting or self.is_connected) then
+        return "no connection"
+    end
+
+    if not topic or #topic == 0 then
+        return "topic is required"
+    end
+    if not payload then
+        payload = ""
+    end
+
+    local flags = 0
+    local length = 3 + #topic + #payload
+    local data = string.char(0x30 | flags) .. encodeVarint(length) .. string.pack("> s2 B", topic, 0) .. payload
+
+    local _, err = self.conn:write(data)
+    if err ~= nil then
+        return err
+    end
+
+    local _, err = self.conn:flush()
+    if err ~= nil then
+        return err
+    end
+
+    return nil
+end
+
 function MqttClient:disconnect (reason)
     if not (self.is_connecting or self.is_connected) then
-        return nil
+        return "no connection"
     end
 
     if reason == nil then