Persist-sqlite: add flush_period and page_size options

Signed-off-by: Abilio Marques <abiliojr@gmail.com>

plugins/persist-sqlite/init.c

@@ -268,12 +268,13 @@ int persist_sqlite__init(struct mosquitto_sqlite *ms)
 				ms->db_file, sqlite3_errstr(rc));
 		return MOSQ_ERR_UNKNOWN;
 	}
+	snprintf(buf, sizeof(buf), "PRAGMA page_size=%u;", ms->page_size);
+	rc = sqlite3_exec(ms->db, buf, NULL, NULL, NULL);
+	if(rc) goto fail;
 	rc = sqlite3_exec(ms->db, "PRAGMA journal_mode=WAL;", NULL, NULL, NULL);
 	if(rc) goto fail;
 	rc = sqlite3_exec(ms->db, "PRAGMA foreign_keys = ON;", NULL, NULL, NULL);
 	if(rc) goto fail;
-	rc = sqlite3_exec(ms->db, "PRAGMA page_size=32768;", NULL, NULL, NULL);
-	if(rc) goto fail;
 	snprintf(buf, sizeof(buf), "PRAGMA synchronous=%d;", ms->synchronous);
 	rc = sqlite3_exec(ms->db, buf, NULL, NULL, NULL);
 	if(rc) goto fail;

plugins/persist-sqlite/persist_sqlite.h

@@ -47,7 +47,9 @@ struct mosquitto_sqlite {
 	sqlite3_stmt *retain_msg_remove_stmt;
 	time_t last_transaction;
 	int synchronous;
-	int event_count;
+	unsigned int event_count;
+	unsigned int flush_period;
+	unsigned int page_size;
 };
 
 int persist_sqlite__init(struct mosquitto_sqlite *ms);

plugins/persist-sqlite/plugin.c

@@ -34,6 +34,29 @@ Contributors:
 static mosquitto_plugin_id_t *plg_id = NULL;
 static struct mosquitto_sqlite plg_data;
 
+static int conf_parse_uint(const char *in, const char *name, unsigned int *value)
+{
+	int v = atoi(in);
+	if(v <= 0){
+		mosquitto_log_printf(MOSQ_LOG_ERR, "Error: Invalid '%s' value in configuration.", name);
+		return MOSQ_ERR_INVAL;
+	}
+
+	*value = v;
+	return MOSQ_ERR_SUCCESS;
+}
+
+static void set_defaults(void)
+{
+	/* "normal" synchronous mode. */
+	plg_data.synchronous = 1;
+
+	/* 5 seconds */
+	plg_data.flush_period = 5;
+
+	plg_data.page_size = 4 * 1024;
+}
+
 int mosquitto_plugin_version(int supported_version_count, const int *supported_versions)
 {
 	int i;
@@ -54,8 +77,7 @@ int mosquitto_plugin_init(mosquitto_plugin_id_t *identifier, void **user_data, s
 	UNUSED(user_data);
 
 	memset(&plg_data, 0,sizeof(struct mosquitto_sqlite));
-	/* Default to "normal" synchronous mode. */
+	set_defaults();
-	plg_data.synchronous = 1;
 
 	for(i=0; i<option_count; i++){
 		if(!strcasecmp(options[i].key, "db_file")){
@@ -63,7 +85,6 @@ int mosquitto_plugin_init(mosquitto_plugin_id_t *identifier, void **user_data, s
 			if(plg_data.db_file == NULL){
 				return MOSQ_ERR_NOMEM;
 			}
-			break;
 		}else if(!strcasecmp(options[i].key, "sync")){
 			if(!strcasecmp(options[i].value, "extra")){
 				plg_data.synchronous = 3;
@@ -77,6 +98,12 @@ int mosquitto_plugin_init(mosquitto_plugin_id_t *identifier, void **user_data, s
 				mosquitto_log_printf(MOSQ_LOG_ERR, "Sqlite persistence: Invalid plugin_opt_sync value '%s'.", options[i].value);
 				return MOSQ_ERR_INVAL;
 			}
+		}else if(!strcasecmp(options[i].key, "flush_period")){
+			rc = conf_parse_uint(options[i].value, "flush_period", &plg_data.flush_period);
+			if(rc) return rc;
+		}else if(!strcasecmp(options[i].key, "page_size")){
+			rc = conf_parse_uint(options[i].value, "page_size", &plg_data.page_size);
+			if(rc) return rc;
 		}
 	}
 	if(plg_data.db_file == NULL){

plugins/persist-sqlite/test.conf

@@ -1,2 +1,4 @@
 plugin ./mosquitto_persist_sqlite.so
 plugin_opt_db_file test.sqlite3
+plugin_opt_page_size 4096
+plugin_opt_flush_period 5

plugins/persist-sqlite/tick.c

@@ -31,8 +31,7 @@ int persist_sqlite__tick_cb(int event, void *event_data, void *userdata)
 	struct mosquitto_sqlite *ms = userdata;
 
 	UNUSED(event);
-
+	if(ed->now_s > ms->last_transaction + ms->flush_period && ms->event_count > 0){
-	if(ed->now_s > ms->last_transaction + 5 && ms->event_count > 0){
 		ms->last_transaction = ed->now_s;
 		ms->event_count = 0;
 		sqlite3_exec(ms->db, "END;", NULL, NULL, NULL);