tmpo daemaon with query callback

openwrt/package/flukso/luasrc/tmpod.lua

@@ -39,7 +39,8 @@ local DEBUG = {
 	config = false,
 	block8 = false,
 	compact = false,
-	sync = false
+	sync = false,
+	query = false
 }
 
 local DAEMON = os.getenv("DAEMON") or "tmpod"
@@ -70,15 +71,16 @@ local TMPO_REGEX_V1 = '^,"v":%[0(.*)$'
 local TMPO_REGEX_V2 = '^(.-)%].*$'
 local TMPO_REGEX_SYNC = "^/d/device/(%x+)/tmpo/sync$"
 local TMPO_REGEX_SENSOR = "^/sensor/(%x+)/(%l+)$"
-local TMPO_REGEX_QUERY = "^/query/(%x+)/tmpo$"
 local TMPO_TOPIC_SYNC_SUB = "/d/device/%s/tmpo/sync"
 local TMPO_TOPIC_SYNC_PUB = "/device/%s/tmpo/sync"
 local TMPO_TOPIC_SENSOR_SUB = "/sensor/+/+"
 local TMPO_TOPIC_SENSOR_PUB = "/sensor/%s/tmpo/%d/%d/%d/gz"
 -- /sensor/[sid]/tmpo/[rid]/[lvl]/[bid]/gz
 
+local TMPO_REGEX_QUERY = "^/query/(%x+)/tmpo$"
 local TMPO_TOPIC_QUERY_PUB = "/sensor/%s/query" -- provide queried data as payload
 local TMPO_TOPIC_QUERY_SUB = "/query/+/tmpo" -- get sensor to query with payload interval
+local TMPO_FMT_QUERY = "time:%d sid:%s rid:%d lvl:%2d bid:%d"
 
 local TMPO_GC20_THRESHOLD = 100 -- 100 free 4kB blocks out of +-1000 in jffs2 = 90% full
 local TMPO_GZCHECK_EXEC_FMT = "gzip -trS '' %s 2>&1"
@@ -94,6 +96,7 @@ local MOSQ_TIMEOUT = 0 -- return instantly from select call
 local MOSQ_MAX_PKTS = 1 -- packets
 local MOSQ_QOS0 = 0
 local MOSQ_QOS1 = 1
+local MOSQ_QOS2 = 2
 local MOSQ_RETAIN = true
 local MOSQ_ERROR = "MQTT error: %s"
 
@@ -614,6 +617,27 @@ mqtt:set_callback(mosq.ON_MESSAGE, function(mid, topic, jpayload, qos, retain)
 		return files
 	end
 
+        local function dprint(fmt, ...)
+                if DEBUG.query then
+                        print(fmt:format(
+                                os.time(),
+                        ...))
+                end
+        end
+
+        local function publish(sid, rid, lvl, bid)
+                dprint(TMPO_FMT_QUERY, sid, rid, lvl, bid)
+                local path = TMPO_PATH_TPL:format(sid, rid, lvl, bid)
+                local source = assert(io.open(path, "r"))
+                local payload = source:read("*all")
+                local topic = TMPO_TOPIC_QUERY_PUB:format(sid)
+                if DEBUG.query then
+                        print("publishing", topic, payload)
+                end
+                mqtt:publish(topic, payload, MOSQ_QOS2, not MOSQ_RETAIN)
+                source:close()
+        end
+
 	local function sensor(sid, dtype)
 		local sparams = config.sensor[sid]
 		if not (sid and sparams and dtype == sparams.data_type) then return end
@@ -629,23 +653,30 @@ mqtt:set_callback(mosq.ON_MESSAGE, function(mid, topic, jpayload, qos, retain)
 		tmpo:sync1(payload)
 	end
 
--- publish the stored files on a query request
-	local function query(sid)
--- payload contains query time interval {from:fromtimestamp, to:totimestamp}
-        	local payload = luci.json.decode(jpayload)
-		for rid in nixio.fs.dir(TMPO_BASE_PATH .. sid) do
-			for _, lvl in ipairs(sdir(TMPO_PATH_TPL:format(sid, rid, "", ""))) do
-				for _, bid in ipairs(sdir(TMPO_PATH_TPL:format(sid, rid, lvl, ""))) do
-					if bid >= payload.from and bid <= payload.to
-						-- publish the respective file containing the requested values
-						-- note, the query interval may be smaller than a file's content
-						-- then the respective file must be sent
-					end
-				end
-			end
-		end
-        	return true
-	end
+        -- publish the stored files on a query request
+        local function query(sid)
+                -- payload contains query time interval [fromtimestamp, totimestamp]
+                local payload = luci.json.decode(jpayload)
+                local lastbid = 0
+                if DEBUG.query then
+                        print("entered query with ", sid, payload[1], payload[2])
+                end
+                for rid in nixio.fs.dir(TMPO_BASE_PATH .. sid) do
+                        for _, lvl in ipairs(sdir(TMPO_PATH_TPL:format(sid, rid, "", ""))) do
+                                for _, bid in ipairs(sdir(TMPO_PATH_TPL:format(sid, rid, lvl, ""))) do
+                                        -- detect store with containing or overlapping values
+                                        if ((payload[1] <= bid) and (bid <= payload[2])) then
+                                                publish(sid, rid, lvl, bid)
+                                        end
+                                        if ((lastbid ~= 0) and (bid >= payload[2]) and (lastbid <= payload[1])) then
+                                                publish(sid, rid, lvl, lastbid)
+                                        end
+                                        lastbid = bid
+                                end
+                        end
+                end
+                return true
+        end
 
 	if retain then return end
 	if not sensor(topic:match(TMPO_REGEX_SENSOR)) then