Improved version of luamqtt with fewer bugs

controller-host/mqtt/connector/base/buffered_base.lua

@@ -0,0 +1,89 @@
+-- base connector class for buffered reading.
+--
+-- Use this base class if the sockets do NOT yield.
+-- So LuaSocket for example, when using Copas or OpenResty
+-- use the non-buffered base class.
+--
+-- This base class derives from `non_buffered_base` it implements the
+-- `receive` and `buffer_clear` methods. But adds the `plain_receive` method
+-- that must be implemented.
+--
+-- NOTE: the `plain_receive` method is supposed to be non-blocking (see its
+-- description), but the `send` method has no such facilities, so is `blocking`
+-- in this class. Make sure to set the proper timeouts in either method before
+-- starting the send/receive. So for example for LuaSocket call `settimeout(0)`
+-- before receiving, and `settimeout(30)` before sending.
+--
+-- @class mqtt.connector.base.buffered_base
+
+
+local super = require "mqtt.connector.base.non_buffered_base"
+local buffered = setmetatable({}, super)
+buffered.__index = buffered
+buffered.super = super
+buffered.type = "buffered, blocking i/o"
+
+-- debug helper function
+-- function buffered:buffer_state(msg)
+-- 	print(string.format("buffer: size = %03d  last-byte-done = %03d -- %s",
+-- 			#(self.buffer_string or ""), self.buffer_pointer or 0, msg))
+-- end
+
+-- bytes read were handled, clear those
+function buffered:buffer_clear()
+	-- self:buffer_state("before clearing buffer")
+	self.buffer_string = nil
+	self.buffer_pointer = nil
+end
+
+-- read bytes, first from buffer, remaining from function
+-- if function returns "idle" then reset read pointer
+function buffered:receive(size)
+	-- self:buffer_state("receive start "..size.." bytes")
+
+	local buf = self.buffer_string or ""
+	local idx = self.buffer_pointer or 0
+
+	while size > (#buf - idx) do
+		-- buffer is lacking bytes, read more...
+		local data, err = self:plain_receive(size - (#buf - idx))
+		if not data then
+			if err == self.signal_idle then
+				-- read timedout, retry entire packet later, reset buffer
+				self.buffer_pointer = 0
+			end
+			return data, err
+		end
+
+		-- append received data, and try again
+		buf = buf .. data
+		self.buffer_string = buf
+		-- self:buffer_state("receive added "..#data.." bytes")
+	end
+
+	self.buffer_pointer = idx + size
+	local data = buf:sub(idx + 1, idx + size)
+	-- print("data: ", require("mqtt.tools").hex(data))
+	-- self:buffer_state("receive done "..size.." bytes\n")
+	return data
+end
+
+--- Retrieves the requested number of bytes from the socket, in a non-blocking
+-- manner.
+-- The implementation MUST read with a timeout that immediately returns if there
+-- is no data to read. If there is no data, then it MUST return
+-- `nil, self.signal_idle` to indicate it no data was there and we need to retry later.
+--
+-- If there is partial data, it should return that data (less than the requested
+-- number of bytes), with no error/signal.
+--
+-- If the receive errors, because of a closed connection it should return
+-- `nil, self.signal_closed` to indicate this. Any other errors can be returned
+-- as a regular `nil, err`.
+-- @tparam size int number of bytes to receive.
+-- @return data, or `false, err`, where `err` can be a signal.
+function buffered:plain_receive(size) -- luacheck: ignore
+    error("method 'plain_receive' on buffered connector wasn't implemented")
+end
+
+return buffered

controller-host/mqtt/connector/base/luasec.lua

@@ -0,0 +1,29 @@
+-- validates the LuaSec options, and applies defaults
+return function(conn)
+	if conn.secure then
+		local params = conn.secure_params
+		if not params then
+			-- set default LuaSec options
+			conn.secure_params = {
+				mode = "client",
+				protocol = "any",
+				verify = "none",
+				options = {"all", "no_sslv2", "no_sslv3", "no_tlsv1"},
+			}
+			return
+		end
+
+		local ok, ssl = pcall(require, conn.ssl_module)
+		assert(ok, "ssl_module '"..tostring(conn.ssl_module).."' not found, secure connections unavailable")
+
+		assert(type(params) == "table", "expecting .secure_params to be a table, got: "..type(params))
+
+		params.mode = params.mode or "client"
+		assert(params.mode == "client", "secure parameter 'mode' must be set to 'client' if given, got: "..tostring(params.mode))
+
+		local ctx, err = ssl.newcontext(params)
+		if not ctx then
+			error("Couldn't create secure context: "..tostring(err))
+		end
+	end
+end

controller-host/mqtt/connector/base/non_buffered_base.lua

@@ -0,0 +1,67 @@
+-- base connector class for non-buffered reading.
+--
+-- Use this base class if the sockets DO yield.
+-- So Copas or OpenResty for example, when using LuaSocket
+-- use the buffered base class.
+--
+-- NOTE: when the send operation can also yield (as is the case with Copas and
+-- OpenResty) you should wrap the `send` handler in a lock to prevent a half-send
+-- message from being interleaved by another message send from another thread.
+--
+-- @class mqtt.connector.base.non_buffered_base
+
+
+local non_buffered = {
+	type = "non-buffered, yielding i/o",
+	timeout = 30 -- default timeout
+}
+non_buffered.__index = non_buffered
+
+-- we need to specify signals for these conditions such that the client
+-- doesn't have to rely on magic strings like "timeout", "wantread", etc.
+-- the connector is responsible for translating those connector specific
+-- messages to a generic signal
+non_buffered.signal_idle = {} -- read timeout occured, so we're idle need to come back later and try again
+non_buffered.signal_closed = {}	-- remote closed the connection
+
+--- Validate connection options.
+function non_buffered:shutdown() -- luacheck: ignore
+    error("method 'shutdown' on connector wasn't implemented")
+end
+
+--- Clears consumed bytes.
+-- Called by the mqtt client when the consumed bytes from the buffer are handled
+-- and can be cleared from the buffer.
+-- A no-op for the non-buffered classes, since the sockets yield when incomplete.
+function non_buffered.buffer_clear()
+end
+
+--- Retrieves the requested number of bytes from the socket.
+-- If the receive errors, because of a closed connection it should return
+-- `nil, self.signal_closed` to indicate this. Any other errors can be returned
+-- as a regular `nil, err`.
+-- @tparam size int number of retrieve to return.
+-- @return data, or `false, err`, where `err` can be a signal.
+function non_buffered:receive(size) -- luacheck: ignore
+    error("method 'receive' on non-buffered connector wasn't implemented")
+end
+
+--- Open network connection to `self.host` and `self.port`.
+-- @return `true` on success, or `false, err` on failure
+function non_buffered:connect() -- luacheck: ignore
+    error("method 'connect' on connector wasn't implemented")
+end
+
+--- Shutdown the network connection.
+function non_buffered:shutdown() -- luacheck: ignore
+    error("method 'shutdown' on connector wasn't implemented")
+end
+
+--- Shutdown the network connection.
+-- @tparam data string data to send
+-- @return `true` on success, or `false, err` on failure
+function non_buffered:send(data) -- luacheck: ignore
+    error("method 'send' on connector wasn't implemented")
+end
+
+return non_buffered

controller-host/mqtt/connector/copas.lua

@@ -0,0 +1,121 @@
+--- Copas based connector.
+--
+-- Copas is an advanced coroutine scheduler in pure-Lua. It uses LuaSocket
+-- under the hood, but in a non-blocking way. It also uses LuaSec for TLS
+-- based connections (like the `mqtt.connector.luasocket` one). And hence uses
+-- the same defaults for the `secure` option when creating the `client`.
+--
+-- Caveats:
+--
+-- * the `client` option `ssl_module` is not supported by the Copas connector,
+--   It will always use the module named `ssl`.
+--
+-- * multiple threads can send simultaneously (sending is wrapped in a lock)
+--
+-- * since the client creates a long lived connection for reading, it returns
+--   upon receiving a packet, to call an event handler. The handler must return
+--   swiftly, since while the handler runs the socket will not be reading.
+--   Any task that might take longer than a few milliseconds should be off
+--   loaded to another thread (the Copas-loop will take care of this).
+--
+-- NOTE: you will need to install copas like this: `luarocks install copas`.
+-- @module mqtt.connector.copas
+
+local super = require "mqtt.connector.base.non_buffered_base"
+local connector = setmetatable({}, super)
+connector.__index = connector
+connector.super = super
+
+local socket = require("socket")
+local copas = require("copas")
+local new_lock = require("copas.lock").new
+local validate_luasec = require("mqtt.connector.base.luasec")
+
+
+-- validate connection options
+function connector:validate()
+	if self.secure then
+		assert(self.ssl_module == "ssl" or self.ssl_module == nil, "Copas connector only supports 'ssl' as 'ssl_module'")
+
+		validate_luasec(self)
+	end
+end
+
+-- 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()
+	self:validate()
+	local sock = copas.wrap(socket.tcp(), self.secure_params)
+	copas.setsocketname("mqtt@"..self.host..":"..self.port, sock)
+
+	sock:settimeouts(self.timeout, self.timeout, -1) -- no timout on reading
+
+	local ok, err = sock:connect(self.host, self.port)
+	if not ok then
+		return false, "copas.connect failed: "..err
+	end
+	self.sock = sock
+	self.send_lock = new_lock(30) -- 30 second timeout
+	return true
+end
+
+-- the packet was fully read, we can clear the bufer.
+function connector:buffer_clear()
+	-- since the packet is complete, we wait now indefinitely for the next one
+	self.sock:settimeouts(nil, nil, -1) -- no timeout on reading
+end
+
+-- Shutdown network connection
+function connector:shutdown()
+	self.sock:close()
+	self.send_lock:destroy()
+end
+
+-- Send data to network connection
+function connector:send(data)
+	-- cache locally in case lock/sock gets replaced while we were sending
+	local sock = self.sock
+	local lock = self.send_lock
+
+	local ok, err = lock:get()
+	if not ok then
+		return nil, "failed acquiring send_lock: "..tostring(err)
+	end
+
+	local i = 1
+	while i < #data do
+		i, err = sock:send(data, i)
+		if not i then
+			lock:release()
+			return false, err
+		end
+	end
+	lock:release()
+	return true
+end
+
+-- Receive given amount of data from network connection
+function connector:receive(size)
+	local sock = self.sock
+	local data, err = sock:receive(size)
+	if data then
+		-- bytes received, so change from idefinite timeout to regular until
+		-- packet is complete (see buffer_clear method)
+		self.sock:settimeouts(nil, nil, self.timeout)
+		return data
+	end
+
+	if err == "closed" then
+		return false, self.signal_closed
+	elseif err == "timout" then
+		return false, self.signal_idle
+	else
+		return false, err
+	end
+end
+
+-- export module table
+return connector
+
+-- vim: ts=4 sts=4 sw=4 noet ft=lua

controller-host/mqtt/connector/init.lua

@@ -0,0 +1,34 @@
+--- Auto detect the connector to use.
+-- The different environments require different socket implementations to work
+-- properly. The 'connectors' are an abstraction to facilitate that without
+-- having to modify the client itself.
+--
+-- This module is will auto-detect the environment and return the proper
+-- module from;
+--
+-- * `mqtt.connector.nginx` for using the non-blocking OpenResty co-socket apis
+--
+-- * `mqtt.connector.copas` for the non-blocking Copas wrapped sockets
+--
+-- * `mqtt.connector.luasocket` for LuaSocket based sockets (blocking)
+--
+-- Since the selection is based on a.o. packages loaded, make sure that in case
+-- of using the `copas` scheduler, you require it before the `mqtt` modules.
+--
+-- Since the `client` defaults to this module (`mqtt.connector`) there typically
+-- is no need to use this directly. When implementing your own connectors,
+-- the included connectors provide good examples of what to look out for.
+-- @module mqtt.connector
+
+local loops = setmetatable({
+	copas = "mqtt.connector.copas",
+	nginx = "mqtt.connector.nginx",
+	ioloop = "mqtt.connector.luasocket"
+}, {
+	__index = function()
+		error("failed to auto-detect connector to use, please set one explicitly", 2)
+	end
+})
+local loop = require("mqtt.loop.detect")()
+
+return require(loops[loop])

controller-host/mqtt/connector/luasocket.lua

@@ -0,0 +1,142 @@
+--- LuaSocket (and LuaSec) based connector.
+--
+-- This connector works with the blocking LuaSocket sockets. This connector uses
+-- `LuaSec` for TLS connections. This is the connector used for the included
+-- `mqtt.ioloop` scheduler.
+--
+-- When using TLS / MQTTS connections, the `secure` option passed to the `client`
+-- when creating it, can be the standard table of options as used by LuaSec
+-- for creating a context. When omitted the defaults will be;
+-- `{ mode="client", protocol="any", verify="none",
+--  options={ "all", "no_sslv2", "no_sslv3", "no_tlsv1" } }`
+--
+-- Caveats:
+--
+-- * since the client creates a long lived connection for reading, it returns
+--   upon receiving a packet, to call an event handler. The handler must return
+--   swiftly, since while the handler runs the socket will not be reading.
+--   Any task that might take longer than a few milliseconds should be off
+--   loaded to another task.
+--
+-- @module mqtt.connector.luasocket
+
+local super = require "mqtt.connector.base.buffered_base"
+local luasocket = setmetatable({}, super)
+luasocket.__index = luasocket
+luasocket.super = super
+
+local socket = require("socket")
+local validate_luasec = require("mqtt.connector.base.luasec")
+
+
+-- table with error messages that indicate a read timeout
+luasocket.timeout_errors = {
+	timeout = true,   -- luasocket
+	wantread = true,  -- luasec
+	wantwrite = true, -- luasec
+}
+
+-- validate connection options
+function luasocket:validate()
+	if self.secure then
+		validate_luasec(self)
+	end
+end
+
+-- 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 luasocket:connect()
+	self:validate()
+
+	local ssl
+	if self.secure then
+		ssl = require(self.ssl_module)
+	end
+
+	self:buffer_clear()  -- sanity
+	local sock = socket.tcp()
+	sock:settimeout(self.timeout)
+
+	local ok, err = sock:connect(self.host, self.port)
+	if not ok then
+		return false, "socket.connect failed to connect to '"..tostring(self.host)..":"..tostring(self.port).."': "..err
+	end
+
+	if self.secure_params then
+		-- Wrap socket in TLS one
+		do
+			local wrapped
+			wrapped, err = ssl.wrap(sock, self.secure_params)
+			if not wrapped then
+				sock:close(self) -- close TCP level
+				return false, "ssl.wrap() failed: "..tostring(err)
+			end
+			-- replace sock with wrapped secure socket
+			sock = wrapped
+		end
+
+		-- do TLS/SSL initialization/handshake
+		sock:settimeout(self.timeout) -- sanity; again since its now a luasec socket
+		ok, err = sock:dohandshake()
+		if not ok then
+			sock:close()
+			return false, "ssl dohandshake failed: "..tostring(err)
+		end
+	end
+
+	self.sock = sock
+	return true
+end
+
+-- Shutdown network connection
+function luasocket:shutdown()
+	self.sock:close()
+end
+
+-- Send data to network connection
+function luasocket:send(data)
+	local sock = self.sock
+	local i = 0
+	local err
+
+	sock:settimeout(self.timeout)
+
+	while i < #data do
+		i, err = sock:send(data, i + 1)
+		if not i then
+			return false, err
+		end
+	end
+
+	return true
+end
+
+-- Receive given amount of data from network connection
+function luasocket:plain_receive(size)
+	local sock = self.sock
+
+	sock:settimeout(0)
+
+	local data, err, partial = sock:receive(size)
+
+	data = data or partial or ""
+	if #data > 0 then
+		return data
+	end
+
+	-- convert error to signal if required
+	if self.timeout_errors[err or -1] then
+		return false, self.signal_idle
+	elseif err == "closed" then
+		return false, self.signal_closed
+	else
+		return false, err
+	end
+end
+
+
+-- export module table
+return luasocket
+
+-- vim: ts=4 sts=4 sw=4 noet ft=lua

controller-host/mqtt/connector/nginx.lua

@@ -0,0 +1,102 @@
+--- Nginx OpenResty co-sockets based connector.
+--
+-- This connector works with the non-blocking openresty sockets. Note that the
+-- secure setting haven't been implemented yet. It will simply use defaults
+-- when doing a TLS handshake.
+--
+-- Caveats:
+--
+-- * sockets cannot cross phase/context boundaries. So all client interaction
+--   must be done from the timer context in which the client threads run.
+--
+-- * multiple threads cannot send simultaneously (simple scenarios will just
+--   work)
+--
+-- * since the client creates a long lived connection for reading, it returns
+--   upon receiving a packet, to call an event handler. The handler must return
+--   swiftly, since while the handler runs the socket will not be reading.
+--   Any task that might take longer than a few milliseconds should be off
+--   loaded to another thread.
+--
+-- * Nginx timers should be short lived because memory is only released after
+--   the context is destroyed. In this case we're using the fro prolonged periods
+--   of time, so be aware of this and implement client restarts if required.
+--
+-- thanks to @irimiab: https://github.com/xHasKx/luamqtt/issues/13
+-- @module mqtt.connector.nginx
+
+local super = require "mqtt.connector.base.non_buffered_base"
+local ngxsocket = setmetatable({}, super)
+ngxsocket.__index = ngxsocket
+ngxsocket.super = super
+
+-- load required stuff
+local ngx_socket_tcp = ngx.socket.tcp
+local long_timeout = 7*24*60*60*1000 -- one week
+
+-- validate connection options
+function ngxsocket:validate()
+	if self.secure then
+		assert(self.ssl_module == "ssl", "specifying custom ssl module when using Nginx connector is not supported")
+		assert(self.secure_params == nil or type(self.secure_params) == "table", "expecting .secure_params to be a table if given")
+		-- TODO: validate nginx stuff
+	end
+end
+
+-- 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 ngxsocket:connect()
+	-- TODO: add a lock for sending to prevent multiple threads from writing to
+	-- the same socket simultaneously (see the Copas connector)
+	local sock = ngx_socket_tcp()
+	-- set read-timeout to 'nil' to not timeout at all
+	sock:settimeouts(self.timeout * 1000, self.timeout * 1000, long_timeout) -- no timeout on reading
+	local ok, err = sock:connect(self.host, self.port)
+	if not ok then
+		return false, "socket:connect failed: "..err
+	end
+	if self.secure then
+		sock:sslhandshake()
+	end
+	self.sock = sock
+	return true
+end
+
+-- Shutdown network connection
+function ngxsocket:shutdown()
+	self.sock:close()
+end
+
+-- Send data to network connection
+function ngxsocket:send(data)
+	return self.sock:send(data)
+end
+
+function ngxsocket:buffer_clear()
+	-- since the packet is complete, we wait now indefinitely for the next one
+	self.sock:settimeouts(self.timeout * 1000, self.timeout * 1000, long_timeout) -- no timeout on reading
+end
+
+-- Receive given amount of data from network connection
+function ngxsocket:receive(size)
+	local sock = self.sock
+	local data, err = sock:receive(size)
+	if data then
+		-- bytes received, so change from idefinite timeout to regular until
+		-- packet is complete (see buffer_clear method)
+		self.sock:settimeouts(self.timeout * 1000, self.timeout * 1000, self.timeout * 1000)
+		return data
+	end
+
+	if err == "closed" then
+		return false, self.signal_closed
+	elseif err == "timout" then
+		return false, self.signal_idle
+	else
+		return false, err
+	end
+end
+
+-- export module table
+return ngxsocket