Merge pull request #27 from xHasKx/dvv-patch-1

Dvv patch 1 - copas support

luamqtt-3.3.2-1.rockspec → luamqtt-3.4.1-1.rockspec

@@ -1,8 +1,8 @@
 package = "luamqtt"
-version = "3.3.2-1"
+version = "3.4.1-1"
 source = {
 	url = "git://github.com/xHasKx/luamqtt",
-	tag = "v3.3.2",
+	tag = "v3.4.1",
 }
 description = {
 	summary = "luamqtt - Pure-lua MQTT v3.1.1 and v5.0 client",

mqtt/client.lua

@@ -908,7 +908,12 @@ function client_mt:_ioloop_iteration()
 	if conn then
 		-- network connection opened
 		-- perform packet receiving using ioloop receive function
-		local ok, err = self:_io_iteration(ioloop_recv)
+		local ok, err
+		if loop then
+			ok, err = self:_io_iteration(ioloop_recv)
+		else
+			ok, err = self:_sync_iteration()
+		end
 
 		if ok then
 			-- send PINGREQ if keep_alive interval is reached
@@ -933,15 +938,19 @@ function client_mt:_ioloop_iteration()
 						self.reconnect_timer_start = nil
 						self:start_connecting()
 					else
-						loop:can_sleep()
+						if loop then
+							loop:can_sleep()
+						end
 					end
 				else
 					self.reconnect_timer_start = os_time()
 				end
 			end
 		else
 			-- finish working with client
-			loop:remove(self)
+			if loop then
+				loop:remove(self)
+			end
 		end
 	end
 end

mqtt/init.lua

@@ -24,7 +24,7 @@ local mqtt = {
 	v50 = 5,		-- supported protocol version, MQTT v5.0
 
 	-- mqtt library version
-	_VERSION = "3.3.2",
+	_VERSION = "3.4.1",
 }
 
 -- load required stuff

mqtt/luasocket-copas.lua

@@ -0,0 +1,48 @@
+-- DOC: https://keplerproject.github.io/copas/
+-- NOTE: you will need to install copas like this: luarocks install copas
+
+-- module table
+local connector = {}
+
+local socket = require("socket")
+local copas = require("copas")
+
+-- Open network connection to .host and .port in conn table
+-- Store opened socket to conn table
+-- Returns true on success, or false and error text on failure
+function connector.connect(conn)
+	local sock, err = socket.connect(conn.host, conn.port)
+	if not sock then
+		return false, "socket.connect failed: "..err
+	end
+	conn.sock = sock
+	return true
+end
+
+-- Shutdown network connection
+function connector.shutdown(conn)
+	conn.sock:shutdown()
+end
+
+-- Send data to network connection
+function connector.send(conn, data, i, j)
+	local ok, err = copas.send(conn.sock, data, i, j)
+	return ok, err
+end
+
+-- Receive given amount of data from network connection
+function connector.receive(conn, size)
+	local ok, err = copas.receive(conn.sock, size)
+	return ok, err
+end
+
+-- Set connection's socket to non-blocking mode and set a timeout for it
+function connector.settimeout(conn, timeout)
+	conn.timeout = timeout
+	conn.sock:settimeout(0)
+end
+
+-- export module table
+return connector
+
+-- vim: ts=4 sts=4 sw=4 noet ft=lua

openwrt/Makefile

@@ -1,7 +1,7 @@
 include $(TOPDIR)/rules.mk
 
 PKG_NAME:=luamqtt
-PKG_VERSION:=3.3.2
+PKG_VERSION:=3.4.1
 PKG_RELEASE:=1
 PKG_MAINTAINER:=Alexander Kiranov <[email protected]>
 

openwrt/make-package-without-openwrt-sources.sh

@@ -50,7 +50,7 @@ rm -rf $PKG_ROOT/usr
 # prepare control.tar.gz
 cat << EOF > $PKG_ROOT/control
 Package: luamqtt
-Version: 3.3.2-1
+Version: 3.4.1-1
 Depends: libc, lua, luasocket, luabitop, luasec
 Source: https://github.com/xHasKx/luamqtt
 SourceName: luamqtt

tests/run-for-all-lua-versions.sh

@@ -26,6 +26,7 @@ for ver in -l5.1 -l5.2 -l5.3 -j2.0 -j2.1; do
 			luarocks install luabitop > /dev/null 2>&1
 		fi
 		luarocks install busted > /dev/null 2>&1
+		luarocks install copas > /dev/null 2>&1
 		if [ -d /usr/lib/x86_64-linux-gnu ]; then
 			# debian-based OS
 			[ -f /etc/lsb-release ] && . /etc/lsb-release

tests/spec/mqtt-client.lua

@@ -504,5 +504,70 @@ describe("no_local flag for subscription: ", function()
 	end)
 end)
 
+describe("copas connector", function()
+	local mqtt = require("mqtt")
+	local copas = require("copas")
+	local prefix = "luamqtt/" .. tostring(math.floor(math.random()*1e13))
+
+	it("test", function()
+		-- NOTE: more about flespi tokens:
+		-- https://flespi.com/kb/tokens-access-keys-to-flespi-platform
+		local flespi_token = "stPwSVV73Eqw5LSv0iMXbc4EguS7JyuZR9lxU5uLxI5tiNM8ToTVqNpu85pFtJv9"
+
+		local client = mqtt.client{
+			uri = "mqtt.flespi.io",
+			clean = true,
+			username = flespi_token,
+			version = mqtt.v50,
+
+			connector = require("mqtt.luasocket-copas"),
+		}
+
+		client:on{
+			connect = function()
+				-- print("client connected")
+				assert(client:subscribe{topic=prefix.."/copas", qos=1, callback=function()
+					-- print("subscribed")
+					copas.sleep(2)
+					assert(client:publish{
+						topic = prefix.."/copas",
+						payload = "copas test",
+						qos = 1,
+					})
+				end})
+			end,
+
+			message = function(msg)
+				assert(client:acknowledge(msg))
+				-- print("received", msg)
+				if msg.topic == prefix.."/copas" and msg.payload == "copas test" then
+					-- print("disconnect")
+					assert(client:disconnect())
+				end
+			end
+		}
+
+		local ticks = 0
+		local mqtt_finished = false
+
+		copas.addthread(function()
+			mqtt.run_sync(client)
+			mqtt_finished = true
+			assert.is_true(ticks > 1, "expecting mqtt client run takes at least one tick")
+		end)
+
+		copas.addthread(function()
+			for _ = 1, 3 do
+				copas.sleep(1)
+				ticks = ticks + 1
+			end
+		end)
+
+		copas.loop()
+
+		assert.is_true(mqtt_finished, "expecting mqtt client to finish its work")
+		assert.is_true(ticks == 3, "expecting 3 ticks")
+	end)
+end)
 
 -- vim: ts=4 sts=4 sw=4 noet ft=lua