Introduce thread management for future reading job

Change-Id: I1d47aef85c4a90d10d7cd5f85c357e9e35664836
Signed-off-by: Romain Forlot <romain.forlot@iot.bzh>

5 files changed, 44 insertions, 28 deletions

diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt
index 4707070..4217668 100644
--- a/src/CMakeLists.txt
+++ b/src/CMakeLists.txt
@@ -23,6 +23,8 @@ PROJECT_TARGET_ADD(harvester)
 	# Define project Targets
 	add_library(${TARGET_NAME} MODULE
 		${TARGET_NAME}.c
+		plugins/tsdb.c
+		plugins/influxdb.c
 	)
 
 	set(OPENAPI_DEF "harvester-apidef" CACHE STRING "name and path to the JSON API definition without extension")
diff --git a/src/harvester.c b/src/harvester.c
index 63e5a6f..945f6e1 100644
--- a/src/harvester.c
+++ b/src/harvester.c
@@ -18,29 +18,28 @@
 #define _GNU_SOURCE
 #include "harvester.h"
 #include "harvester-apidef.h"
-#include <string.h>
-#include <stdio.h>
+#include <pthread.h>
 
 #include "plugins/tsdb.h"
 #include "curl-wrap.h"
 #include "wrap-json.h"
 
-#define DEFAULT_DB "agl-garner"
-#define DEFAULT_DBHOST "localhost"
-#define DEFAULT_DBPORT "8086"
-#define URL_MAXIMUM_LENGTH 2047
+#define ERROR -1
 
 CURL* (*tsdb_write)(const char* host, int port, json_object *metric);
 CURL* (*tsdb_read)(const char* host, int port, json_object *metric);
 void (*curl_cb)(void *closure, int status, CURL *curl, const char *result, size_t size);
-
+pthread_mutex_t db_mutex;
 
 int do_write(struct afb_req req, const char* host, int port, json_object *metric)
 {
 	CURL *curl_request;
 
 	curl_request = tsdb_write(host, port, metric);
+
+	pthread_mutex_lock(&db_mutex);
 	curl_wrap_do(curl_request, curl_cb, &req);
+	pthread_mutex_unlock(&db_mutex);
 
 	return 0;
 }
@@ -72,16 +71,19 @@ void auth(struct afb_req request)
 
 int init()
 {
-	/* Ok 2 int is no needed, 1 is enough but 2 is more lisible. */
-	int db_up = 0, err = 0;
-	err = curl_global_init(CURL_GLOBAL_DEFAULT);
-
-	if (!err)
-		db_up = db_ping();
-	else
+	int tsdb_available = 0;
+	if(curl_global_init(CURL_GLOBAL_DEFAULT) != 0) {
 		AFB_ERROR("Something went wrong initiliazing libcurl. Abort");
+		return ERROR;
+	}
 
-	switch (db_up) {
+	if(pthread_mutex_init(&db_mutex, NULL) != 0) {
+		AFB_ERROR("Something went wrong initiliazing mutex. Abort");
+		return ERROR;
+	}
+
+	tsdb_available = db_ping();
+	switch (tsdb_available) {
 		case INFLUX:
 			tsdb_write = influxdb_write;
 			tsdb_read = influxdb_read;
@@ -89,9 +91,9 @@ int init()
 			break;
 		default:
 			AFB_ERROR("No Time Series Database found. Abort");
-			err = -1;
+			return ERROR;
 			break;
 	}
 
-	return err;
+	return 0;
 }
diff --git a/src/plugins/influxdb.c b/src/plugins/influxdb.c
index 74c1297..17c140d 100644
--- a/src/plugins/influxdb.c
+++ b/src/plugins/influxdb.c
@@ -15,7 +15,12 @@
  * limitations under the License.
  */
 
+#define _GNU_SOURCE
+#include <string.h>
+#include <stdio.h>
+
 #include "tsdb.h"
+#include "wrap-json.h"
 
 int create_database()
 {
@@ -37,7 +42,6 @@ int create_database()
 	}
 
 	curl_easy_cleanup(request);
-	free(post_data);
 
 	if(ret == 0)
 		AFB_NOTICE("Database 'agl-collector' created");
@@ -148,7 +152,7 @@ CURL *make_curl_write_post(const char *url, struct json_object *metric)
 		curl = NULL;
 	}
 	else {
-		for(int i = lpd; i != 0; i--) {
+		for(long unsigned int i = lpd; i != 0; i--) {
 			format_write_query(query, name, source, unit, identity, jv, timestamp);
 			post_data[i] = i == lpd ? NULL : query;
 		}
@@ -169,5 +173,5 @@ CURL *influxdb_write(const char* host, int port, json_object *metric)
 
 	strncat(url, host, strlen(host));
 	strncat(url, ":"DEFAULT_DBPORT"/write?db="DEFAULT_DB, strlen(":"DEFAULT_DBPORT"/write?db="DEFAULT_DB));
-	curl_request = make_curl_write_post(url, metric);
+	return make_curl_write_post(url, metric);
 }
diff --git a/src/plugins/tsdb.c b/src/plugins/tsdb.c
index 20a3e89..f7a8ee2 100644
--- a/src/plugins/tsdb.c
+++ b/src/plugins/tsdb.c
@@ -17,14 +17,6 @@
 
 #include "tsdb.h"
 
-int db_ping()
-{
-	int ret = 0;
-	if(influxdb_ping() == 0) ret = INFLUX;
-
-	return ret;
-}
-
 int influxdb_ping()
 {
 	int ret = 0;
@@ -41,3 +33,11 @@ int influxdb_ping()
 	curl_easy_cleanup(request);
 	return ret;
 }
+
+int db_ping()
+{
+	int ret = 0;
+	if(influxdb_ping() == 0) ret = INFLUX;
+
+	return ret;
+}
diff --git a/src/plugins/tsdb.h b/src/plugins/tsdb.h
index 967bce0..527f538 100644
--- a/src/plugins/tsdb.h
+++ b/src/plugins/tsdb.h
@@ -15,9 +15,17 @@
  * limitations under the License.
  */
 
+#define AFB_BINDING_VERSION 2
+#include <afb/afb-binding.h>
+
 #include <json-c/json.h>
 #include "curl-wrap.h"
 
+#define DEFAULT_DB "agl-garner"
+#define DEFAULT_DBHOST "localhost"
+#define DEFAULT_DBPORT "8086"
+#define URL_MAXIMUM_LENGTH 2047
+
 enum db_available {
 	INFLUX = 1,
 	GRAPHITE = 2,