Added support for scheduled services

CMakeLists.txt

@@ -9,9 +9,11 @@ find_package(spdlog CONFIG REQUIRED)
 find_package(OpenSSL REQUIRED)
 include(FindLua)
 find_package(lua REQUIRED)
+find_package(InfluxDB REQUIRED)
 find_package(eclipse-paho-mqtt-c CONFIG REQUIRED)
+find_package(croncpp CONFIG REQUIRED)
 
-add_executable(thrawn src/main.cpp src src/thrawn_api.cpp src/Service.cpp src/Cache.cpp src/LuaEngine.cpp lib/mqtt++/src/mqtt++.cpp)
-target_link_libraries(thrawn ${LIBS} spdlog::spdlog  ${LUA_LIBRARIES} OpenSSL::SSL eclipse-paho-mqtt-c::paho-mqtt3cs-static)
+add_executable(thrawn src/main.cpp src src/thrawn_api.cpp src/thrawn_api_db_logger.cpp src/Service.cpp src/Cache.cpp src/LuaEngine.cpp lib/mqtt++/src/mqtt++.cpp src/DBLogger.cpp)
+target_link_libraries(thrawn ${LIBS} croncpp::croncpp spdlog::spdlog  InfluxData::InfluxDB ${LUA_LIBRARIES} OpenSSL::SSL eclipse-paho-mqtt-c::paho-mqtt3cs-static)
 target_include_directories(thrawn PRIVATE ${LUA_INCLUDE_DIR})
 install(TARGETS thrawn DESTINATION bin)

src/Cache.cpp

@@ -1,22 +1,19 @@
 #include "Cache.h"
 #include <algorithm>
-std::vector<TopicState> Cache::topic_states(std::vector<std::string>& topics) {
-	std::vector<TopicState> states;
+#include "mqtt++.h"
+#include <iostream>
 
-	for (const auto& topic : topics) {
-		auto hit = std::find_if(cache.begin(), cache.end(), [&](TopicState& cache_state) {
-			return cache_state.topic == topic;
-		});
-		if (hit != cache.end()) {
-			states.push_back(*hit);
+std::optional<std::string> Cache::cached_topic_value(const std::string& topic) {
+	for (const auto& state : m_cache) {
+		if (state.topic == topic) {
+			return state.value;
 		}
 	}
-
-	return states;
+	return std::nullopt;
 }
 
 TopicState Cache::set_topic_value(const std::string& topic, const std::string& value) {
-	for (auto& state : cache) {
+	for (auto& state : m_cache) {
 		if (state.topic == topic) {
 			state.value = value;
 
@@ -24,5 +21,5 @@ TopicState Cache::set_topic_value(const std::string& topic, const std::string& v
 		}
 	}
 
-	return cache.emplace_back(topic, value);
+	return m_cache.emplace_back(topic, value);
 }

src/Cache.h

@@ -3,12 +3,13 @@
 #include <string>
 #include <vector>
 #include "Service.h"
+#include <optional>
 class Cache {
 	private:
-		std::vector<TopicState> cache;
+		std::vector<TopicState> m_cache;
 	public:
-	std::vector<TopicState> topic_states(std::vector<std::string>& topics);
 	TopicState set_topic_value(const std::string& topic, const std::string& value);
+	std::optional<std::string> cached_topic_value(const std::string& topic);
 };
 
 #endif // CACHE_H

src/DBLogger.cpp

@@ -0,0 +1,53 @@
+/*
+auto influxdb = 
+
+	influxdb->write(influxdb::Point{"test"}
+	.addField("value2", "mu")
+	.addTag("host", "localhost")
+	);
+	*/
+#include "DBLogger.h"
+#include "InfluxDBFactory.h"
+#include "spdlog/spdlog.h" 
+//url = http://localhost:8086
+//databse = test
+DBLogger::DBLogger(const std::string& url, const std::string& database) {
+	m_influx = influxdb::InfluxDBFactory::Get(url + "/?db=" + database);
+}
+
+void DBLogger::write_light(const std::string_view room, const std::string_view name, const bool is_on) {
+	try {
+		m_influx->write(influxdb::Point("light")
+					.addTag("room", room)
+					.addTag("name", name)
+					.addField("on", is_on ? 1 : 0));
+	}catch (const std::exception& e) {
+		spdlog::error("Could not perform influx write for '{}' in room '{}'", room, name);
+	}
+	
+}
+
+void DBLogger::write_light_dimmable(const std::string_view room, const std::string_view name, const bool is_on, const int brightness) {
+	try {
+		m_influx->write(influxdb::Point("light")
+						.addTag("room", room)
+						.addTag("name", name)
+						.addField("on", is_on ? 1 : 0)
+						.addField("brightness", brightness));
+	}catch (const std::exception& e) {
+		spdlog::error("Could not perform influx write for '{}' in room '{}'", room, name);
+	}
+}
+
+void DBLogger::write_light_tuneable_white(const std::string_view room, const std::string_view name, const bool is_on, const int brightness, const int color_temperature) {
+	try {
+		m_influx->write(influxdb::Point("light")
+						.addTag("room", room)
+						.addTag("name", name)
+						.addField("on", is_on ? 1 : 0)
+						.addField("brightness", brightness)
+						.addField("color_temp", color_temperature));
+	}catch (const std::exception& e) {
+		spdlog::error("Could not perform influx write for '{}' in room '{}'", room, name);
+	}
+}

src/DBLogger.h

@@ -0,0 +1,20 @@
+#if !defined(DBLOGGER_H)
+#define DBLOGGER_H
+
+#include <string>
+#include "InfluxDB.h"
+#include <memory>
+#include <string_view>
+
+class DBLogger {
+	private:
+	std::unique_ptr<influxdb::InfluxDB> m_influx;
+	public:
+	void write_light(const std::string_view room, const std::string_view name, const bool is_on);
+	void write_light_dimmable(const std::string_view room, const std::string_view name, const bool is_on, const int brightness);
+	void write_light_tuneable_white(const std::string_view room, const std::string_view name, const bool is_on, const int brightness, const int color_temperature);
+
+	DBLogger(const std::string& url, const std::string& database);
+};
+
+#endif // DBLOGGER_H

src/LuaEngine.cpp

@@ -1,8 +1,16 @@
 #include "LuaEngine.h"
+#include <ctime>
 
-LuaEngine::LuaEngine() {
+bool ScheduledService::operator<(const ScheduledService& rhs) const {
+	return next_run > rhs.next_run;
 }
 
+void ScheduledService::next() {
+	next_run = cron::cron_next(cron, std::time(0));  
+}
+
+LuaEngine::LuaEngine() : db_logger("http://localhost:8086", "test") {}
+
 void LuaEngine::load_services() {
 	for (auto& service : services) {
 		service.load();

src/LuaEngine.h

@@ -5,12 +5,31 @@
 #include <vector>
 #include <functional>
 #include <string>
+#include "DBLogger.h"
+#include "croncpp/croncpp.h"
+#include <ctime>
+#include <queue>
+#include "spdlog/spdlog.h"
+
+struct ScheduledService {
+	cron::cronexpr cron;
+	std::time_t next_run;
+	LuaService* service;
+	std::string callback;
+
+	void next();
+	bool operator<(const ScheduledService& rhs) const;
+};
+
 
 class LuaEngine {
 	private:
 		LuaEngine();
 	public:
 		std::function<void(const TopicState& state)> publish;
+		DBLogger db_logger;
+		std::priority_queue<ScheduledService> time_schedule;
+
 	static LuaEngine& get_instance() {
 		static LuaEngine instance;
 
@@ -20,13 +39,51 @@ class LuaEngine {
 	
 	Cache cache;
 
+
 	void load_services();
 	void start_services();
 	LuaService* create_service(std::string filename);
 	LuaService* service_by_lua_state(lua_State* lua);
-
 	LuaEngine(LuaEngine const&) = delete;
 	void operator=(LuaEngine const&)  = delete;
 };
 
+template<typename Arg, typename... Args>
+inline bool check_arguments(lua_State* lua, int stack_pos) {
+    return check_arguments<Args...>(lua, stack_pos+1) && check_arguments<Arg>(lua, stack_pos);
+}
+ 
+template<typename Arg, typename... Args>
+inline bool check_arguments(lua_State* lua) {
+    int num_arg = sizeof...(Args) + 1;
+    return check_arguments<Arg, Args...>(lua, num_arg * -1);
+}
+ 
+template <>
+inline bool check_arguments<std::string>(lua_State* lua, int stack_pos) {
+    return lua_isstring(lua, stack_pos);
+}
+ 
+template <>
+inline bool check_arguments<int>(lua_State* lua, int stack_pos) {
+    return lua_isnumber(lua, stack_pos);
+}
+ 
+template <>
+inline bool check_arguments<bool>(lua_State* lua, int stack_pos) {
+    return lua_isnumber(lua, stack_pos);
+}
+
+template<typename F>
+void lua_func_call(LuaService& service, const std::string& lua_func, F push_params) {
+	lua_getglobal(service.lua, lua_func.c_str());
+
+	int num_arguments = push_params();
+
+	if (lua_pcall(service.lua, num_arguments, 0, 0) != 0) {
+		spdlog::error("[{}] Unable to call '{}'. [Lua: '{}']", lua_func, service.name);
+
+		return;
+	}
+}
 #endif // LUA_ENGINE_H

src/Service.cpp

@@ -48,6 +48,9 @@ LuaService::LuaService(std::string filename) {
 	lua_pushcfunction(lua, thrawn_subscribe);
 	lua_setglobal(lua, "thrawn_subscribe");
 
+	lua_pushcfunction(lua, thrawn_schedule);
+	lua_setglobal(lua, "thrawn_schedule");
+
 	lua_pushcfunction(lua, thrawn_publish);
 	lua_setglobal(lua, "thrawn_publish");
 
@@ -57,8 +60,21 @@ LuaService::LuaService(std::string filename) {
 	lua_pushcfunction(lua, thrawn_topic_state_str);
 	lua_setglobal(lua, "thrawn_topic_state_str");
 
-	lua_pushcfunction(lua, thrawn_topic_cache);
-	lua_setglobal(lua, "thrawn_topic_cache");
+	lua_pushcfunction(lua, thrawn_cached_topic);
+	lua_setglobal(lua, "thrawn_cached_topic");
+
+	lua_pushcfunction(lua, thrawn_topic_components);
+	lua_setglobal(lua, "thrawn_topic_components");
+	
+	lua_pushcfunction(lua, thrawn_log_light);
+	lua_setglobal(lua, "thrawn_log_light");
+
+	lua_pushcfunction(lua, thrawn_log_light_dimmable);
+	lua_setglobal(lua, "thrawn_log_light_dimmable");
+
+	lua_pushcfunction(lua, thrawn_log_light_tw);
+	lua_setglobal(lua, "thrawn_log_light_tw");
+
 
 	
 	if (luaL_loadfile(lua, filename.c_str()) != 0) {
@@ -76,6 +92,8 @@ LuaService::LuaService(LuaService&& service) noexcept {
 	lua = service.lua;
 	service.lua = nullptr;
 	name = std::move(service.name);
+	loaded = false;
+	started = false;
 }
 
 LuaService& LuaService::operator=(LuaService&& service) noexcept {
@@ -88,9 +106,9 @@ LuaService& LuaService::operator=(LuaService&& service) noexcept {
 
 void LuaService::load() {
 	if (!loaded) {
-		if (lua_pcall(lua, 0, 0 ,0) != 0) {
+		if (lua_pcall(lua, 0, 0 ,0) == 0) {
 			loaded = true;
-			
+		}else{
 			spdlog::error("[{}] Unable to call 'lua_pcall()' in load(). [Lua: '{}']", name, lua_tostring(lua,-1));
 		}
 	}

src/Version.h

@@ -2,7 +2,7 @@
 #define VERSION_H
 
 #define THRAWN_MAJOR_VERSION 1
-#define THRAWN_MINOR_VERSION 0
-#define THRAWN_PATCH_VERSION 3
+#define THRAWN_MINOR_VERSION 1
+#define THRAWN_PATCH_VERSION 0
 
 #endif // VERSION_H

src/main.cpp

@@ -11,6 +11,7 @@
 #include <cstring>
 #include <thread>
 #include "mqtt++.h"
+#include <ctime>
 
 std::vector<std::string> lua_scripts_in_folder(const std::string& path_str) {
 	std::vector<std::string> scripts;
@@ -39,7 +40,6 @@ int main(int argc, char* argv[]) {
 	std::string version_topic_value = std::to_string(THRAWN_MAJOR_VERSION) + "." + std::to_string(THRAWN_MINOR_VERSION) + "." + std::to_string(THRAWN_PATCH_VERSION);
 	std::string lastWillConnected = "1";
 
-
 	/*
 	* Parsing arguments
 	* Show version or load path
@@ -69,12 +69,12 @@ int main(int argc, char* argv[]) {
 
 	mqtt.publish(last_will.topic, lastWillConnected);
 	mqtt.publish(version_topic, version_topic_value);
-	mqtt.subscribe("#", [&] (const char* topic, const std::string& msg) {
+	mqtt.subscribe("#", [&] (const std::string& topic, const std::string& msg) {
 		LuaEngine::get_instance().cache.set_topic_value(topic, msg);
 
 		for (auto& service : LuaEngine::get_instance().services) {
 			for (const std::string& sub_topic : service.subscriptions) {
-				if (sub_topic == topic) {
+				if (mqttpp::sub_matches_topic(sub_topic, topic)) {
 					service.on_update(TopicState(topic, msg));
 				}
 			}
@@ -111,6 +111,26 @@ int main(int argc, char* argv[]) {
 	spdlog::info("All services loaded.");
 
 	while (true) {
+		time_t now = std::time(0);
+		
+		bool check_next = true;
+
+		while (check_next) {
+			ScheduledService scheduled_service = LuaEngine::get_instance().time_schedule.top();
+			if (now > scheduled_service.next_run) {
+				LuaEngine::get_instance().time_schedule.pop();
+				call_thrawn_on_schedule(*scheduled_service.service, now, scheduled_service.callback);
+				scheduled_service.next();
+				spdlog::info("Now: {} Next: {}, Time: {}", now, scheduled_service.next_run, std::time(0));
+				LuaEngine::get_instance().time_schedule.push(scheduled_service);
+
+			}else{
+				check_next = false;
+			}
+		}
+		
+		
+
 		using namespace std::chrono_literals;
 		std::this_thread::sleep_for(100ms);
 	}

src/thrawn_api.cpp

@@ -3,6 +3,8 @@
 #include "Cache.h"
 #include "LuaEngine.h"
 #include "spdlog/spdlog.h"
+#include <sstream>
+#include "croncpp/croncpp.h"
 
 /*
 	HELPER
@@ -23,8 +25,8 @@ void push_list_topic_state(lua_State* lua, std::vector<TopicState> states) {
 
 	int i = 1;
 	for (auto& state : states) {
-		lua_pushinteger(lua, i);
-		push_topic_state_table(lua, state);
+		lua_pushstring(lua, state.topic.c_str());
+		lua_pushstring(lua, state.value.c_str());
 		lua_settable(lua, -3);
 
 		i++;
@@ -41,7 +43,7 @@ int thrawn_log(lua_State* lua)
 		return -1;
 	} 
 
-	if (!lua_isstring (lua, -1)) {
+	if (!check_arguments<std::string>(lua)) {
 		return -1;
 	}
 
@@ -135,20 +137,103 @@ int thrawn_subscribe(lua_State* lua) {
 	return 0;
 }
 
-int thrawn_topic_cache(lua_State* lua) {
+int thrawn_schedule(lua_State* lua) {
 	int paramCount = lua_gettop(lua);
 
-	if(paramCount != 0){
+	if(paramCount < 1 || paramCount > 2){
 		return -1;
 	}
 
-	LuaEngine& lua_engine = LuaEngine::get_instance();
-	LuaService* service = lua_engine.service_by_lua_state(lua);
+	LuaService* service = LuaEngine::get_instance().service_by_lua_state(lua);
 	
-	push_list_topic_state(lua, lua_engine.cache.topic_states(service->subscriptions));
+	if (service->name == "") {
+		spdlog::error("Not registered service tried to subscribe to schedule '{}'", lua_tostring(lua, -1));
+
+		return 0;
+	}
+
+	try {
+		ScheduledService scheduled_service;
+		
+		std::string cron_str = lua_tostring(lua, -1);
+		scheduled_service.callback = "thrawn_on_schedule";
+
+		if (paramCount == 2) {
+			cron_str = lua_tostring(lua, -2);
+			scheduled_service.callback = lua_tostring(lua, -1);
+		}
+
+		scheduled_service.cron = cron::make_cron(cron_str);
+		
+		scheduled_service.service = service;
+		scheduled_service.next();
+
+		LuaEngine::get_instance().time_schedule.push(scheduled_service);
+
+		spdlog::info("Service '{}' scheduled to '{}'", service->name, cron_str);
+	}
+	catch (cron::bad_cronexpr const & ex)
+	{
+		spdlog::error("Could not register scheduled job for cron string '{}' and service '{}'", lua_tostring(lua, -1), service->name);
+	}
+
+	
+
+	return 0;
+}
+
+int thrawn_cached_topic(lua_State* lua) {
+	int paramCount = lua_gettop(lua);
+
+	if (paramCount != 1){
+		return -1;
+	}
+
+	if (!lua_isstring(lua, -1)) {
+		return -1;
+	}
+
+	const auto& cached_value = LuaEngine::get_instance().cache.cached_topic_value(lua_tostring(lua, -1));
+
+	if (cached_value) {
+		lua_pushstring(lua, cached_value.value().c_str());
+	}else {
+		lua_pushnil(lua);
+	}
 
 	return 1;
 }
+
+int thrawn_topic_components(lua_State* lua) {
+	int paramCount = lua_gettop(lua);
+
+	if (paramCount != 1) {
+		return -1;
+	}
+
+	if (!lua_isstring(lua, -1)) {
+		return -1;
+	}
+
+	std::vector<std::string> string_components;
+	std::stringstream topic(lua_tostring(lua, -1));
+	std::string token;
+	while (std::getline(topic, token, '/')) {
+		string_components.push_back(token);
+	}
+
+	lua_newtable(lua);
+	int i = 1;
+	for (const std::string& comp : string_components) {
+		lua_pushinteger(lua, i);
+		lua_pushstring(lua, comp.c_str());
+		lua_settable(lua, -3);
+		i++;
+	}
+
+	return 1;
+}
+
 // thrawn_publish(topic, value)
 int thrawn_publish(lua_State* lua) {
 	int paramCount = lua_gettop(lua);
@@ -169,23 +254,23 @@ int thrawn_publish(lua_State* lua) {
 }
 
 void call_thrawn_on_start(LuaService& service) {
-	lua_getglobal(service.lua, "thrawn_on_start");
-
-	if (lua_pcall(service.lua,0, 0, 0) != 0) {
-		spdlog::error("[{}] Unable to call 'thrawn_on_start'. [Lua: '{}']", service.name, lua_tostring(service.lua,-1));
-
-		return;
-	}
+	lua_func_call(service, "thrawn_on_start", [&]() {
+		return 0;
+	});
 }
 
 void call_thrawn_on_update(LuaService& service, TopicState state) {
-	lua_getglobal(service.lua, "thrawn_on_update");
-
+	lua_func_call(service, "thrawn_on_update", [&]() {
 		push_topic_state_table(service.lua, state);
 
-	if (lua_pcall(service.lua, 1, 0, 0) != 0) {
-		spdlog::error("[{}] Unable to call 'thrawn_on_update'. [Lua: '{}']", service.name, lua_tostring(service.lua,-1));
+		return 1;
+	});
+}
 
-		return;
-	}
+void call_thrawn_on_schedule(LuaService& service, std::time_t now, const std::string& callback) {
+	lua_func_call(service, callback.c_str(), [&]() {
+		lua_pushinteger(service.lua, now);
+
+		return 1;
+	});
 }

src/thrawn_api.h

@@ -11,6 +11,7 @@ extern "C"
 #include <vector>
 #include <string>
 #include "Service.h"
+#include <ctime>
 
 //Functions for LUA
 int thrawn_log(lua_State* lua);
@@ -19,17 +20,30 @@ int thrawn_register_service(lua_State* lua);
 
 int thrawn_subscribe(lua_State* lua);
 
+int thrawn_schedule(lua_State* lua);
+
 int thrawn_publish(lua_State* lua);
 
 int thrawn_create_topic_state(lua_State* lua);
 
 int thrawn_topic_state_str(lua_State* lua);
 
-int thrawn_topic_cache(lua_State* lua);
+int thrawn_topic_components(lua_State* lua);
+
+int thrawn_cached_topic(lua_State* lua);
 
 // Function in LUA
 void call_thrawn_on_start(LuaService& service);
 
 void call_thrawn_on_update(LuaService& service, TopicState state);
 
+void call_thrawn_on_schedule(LuaService& service, std::time_t now, const std::string& callback);
+
+//Database logging
+
+int thrawn_log_light(lua_State* lua);
+
+int thrawn_log_light_dimmable(lua_State* lua);
+
+int thrawn_log_light_tw(lua_State* lua);
 #endif // THRAWN_API_H

src/thrawn_api_db_logger.cpp

@@ -0,0 +1,72 @@
+#include "thrawn_api.h"
+#include "LuaEngine.h"
+
+//void write_light(const std::string_view name, const bool is_on);
+int thrawn_log_light(lua_State* lua) {
+	int paramCount = lua_gettop(lua);
+
+	if (paramCount != 3) {
+		return -1;
+	}
+
+	//Param 1: room
+	//Param 2: name
+	//Param 3: is_on
+	if (!check_arguments<std::string, std::string, bool>(lua)) {
+		return -1;
+	}
+	// if (!lua_isstring(lua, -3) || !lua_isstring (lua, -2) || !lua_isboolean (lua, -1)) {
+	// 	return -1;
+	// }
+
+	LuaEngine::get_instance().db_logger.write_light(lua_tostring(lua, -3), lua_tostring(lua, -2), lua_toboolean(lua, -1));
+
+	return 0;
+}
+
+int thrawn_log_light_dimmable(lua_State* lua) {
+	int paramCount = lua_gettop(lua);
+
+	if (paramCount != 4) {
+		return -1;
+	}
+
+	//Param 1: room
+	//Param 2: name
+	//Param 3: is_on
+	//Param 4: brightness
+	if (!check_arguments<std::string, std::string, bool, int>(lua)) {
+		return -1;
+	}
+	// if (!lua_isstring(lua, -4)  | !lua_isstring(lua, -3) || !lua_isboolean(lua, -2) || !lua_isinteger(lua, -1)) {
+	// 	return -1;
+	// }
+
+	LuaEngine::get_instance().db_logger.write_light_dimmable(lua_tostring(lua, -4), lua_tostring(lua, -3), lua_toboolean(lua, -2), lua_tointeger(lua, -1));
+
+	return 0;
+}
+
+int thrawn_log_light_tw(lua_State* lua) {
+	int paramCount = lua_gettop(lua);
+
+	if (paramCount != 5) {
+		return -1;
+	}
+
+	//Param 1: room
+	//Param 2: name
+	//Param 3: is_on
+	//Param 4: brightness
+	//Param 5: color_temperature
+	if (!check_arguments<std::string, std::string, bool, int, int>(lua)) {
+		return -1;
+	}
+	// if (!lua_isstring(lua, -5) || !lua_isstring(lua, -4) || !lua_isboolean(lua, -3) || !lua_isinteger(lua, -2) || !lua_isinteger(lua, -1)) {
+	// 	return -1;
+	// }
+
+	LuaEngine::get_instance().db_logger.write_light_tuneable_white(lua_tostring(lua, -5), lua_tostring(lua, -4), lua_toboolean(lua, -3), lua_tointeger(lua, -2), lua_tointeger(lua, -1));
+
+	return 0;
+}