Clean way to wait for an api's event

Modify the way to wait for an event. Here we call synchronously a verb of the
test api that will end when a timeout expires or when the reception of the event
ends the request
Useless set up  of verbosity to debug for the monitoring API.
"push_after" messages could be caught without setting up this.

Bug-AGL: SPEC-2003
Depends-On: https://gerrit.automotivelinux.org/gerrit/#/c/18583/

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

3 files changed, 122 insertions, 57 deletions

diff --git a/app-controller-submodule b/app-controller-submodule
-Subproject 70e1d98f1118c4785dde75cb64af9272fce1167
+Subproject 0e395a67ff376a8586ac75144bd71c2944acd62
diff --git a/conf.d/controller/lua.d/aft.lua b/conf.d/controller/lua.d/aft.lua
index f5094c2..5b16bf3 100644
--- a/conf.d/controller/lua.d/aft.lua
+++ b/conf.d/controller/lua.d/aft.lua
@@ -23,6 +23,7 @@ lu.LuaUnit:setOutputType('TAP')
 
 _AFT = {
 	exit = {0, code},
+	apiname = nil,
 	context = _ctx,
 	bindingRootDir = nil,
 	tests_list = {},
@@ -32,6 +33,7 @@ _AFT = {
 	afterEach = nil,
 	beforeAll = nil,
 	afterAll = nil,
+	waiting = false,
 	lavaOutput = false,
 }
 
@@ -150,12 +152,26 @@ function _AFT.bindingEventHandler(eventObj, uid)
 	end
 
 	if type(_AFT.monitored_events[eventName]) == 'table' then
+		local stopSync = true
 		if eventListeners then
 			_AFT.monitored_events[eventName].eventListeners = eventListeners
 		end
 
 		_AFT.incrementCount(_AFT.monitored_events[eventName])
 		_AFT.registerData(_AFT.monitored_events[eventName], data)
+
+		for name,value in pairs(_AFT.monitored_events) do
+			if (_AFT.monitored_events[name].expected and
+			   _AFT.monitored_events[name].receivedCount < _AFT.monitored_events[name].expected)
+			then
+				stopSync = false
+			end
+		end
+
+		if stopSync == true and _AFT.waiting == true then
+			AFB:servsync(_AFT.context, _AFT.apiname, "sync", { stop = 1 })
+			_AFT.waiting = false
+		end
 	end
 end
 
@@ -176,55 +192,38 @@ function _AFT.lockWait(eventName, timeout)
 		return 0
 	end
 
-	local count = 0
-	if _AFT.monitored_events[eventName].receivedCount and timeout then
-		count = _AFT.monitored_events[eventName].receivedCount
-	end
+	_AFT.waiting = true
+	local err,responseJ = AFB:servsync(_AFT.context, _AFT.apiname, "sync", { start = timeout})
 
-	while timeout > 0 do
-		timeout = AFB:lockwait(_AFT.context, timeout)
-		AFB:lockwait(_AFT.context, 0) --without it _evt_catcher_ cannot received event
-
-		if _AFT.monitored_events[eventName].receivedCount == count + 1 then
-		return 1
-		end
+	if err then
+		return 0
 	end
-	return 0
+	return 1
 end
 
 function _AFT.lockWaitGroup(eventGroup, timeout)
+	local count = 0
 	if type(eventGroup) ~= "table" then
 		print("Error: wrong argument given to wait a group of events. 1st argument should be a table")
 		return 0
 	end
 
-	-- Copy and compute the expected as it may have already received events
-	-- you should add the expected count to the actual received counter to be
-	-- accurate.
-	local eventGroupCpy = {}
 	for event,expectedCount in pairs(eventGroup) do
-		eventGroupCpy[event] = expectedCount + _AFT.monitored_events[event].receivedCount
+		_AFT.monitored_events[event].expected = expectedCount + _AFT.monitored_events[event].receivedCount
 	end
 
-	local total = 0
-	local matched = nil
-	while timeout > 0 do
-		timeout = AFB:lockwait(_AFT.context, timeout)
-		AFB:lockwait(_AFT.context, 0) --without it _evt_catcher_ cannot received event
+	_AFT.waiting = true
+	local err, responseJ = AFB:servsync(_AFT.context, _AFT.apiname, "sync", { start = timeout })
 
-		for name,expectedCount in pairs(eventGroupCpy) do
-			if _AFT.monitored_events[name].receivedCount >= expectedCount then
-				total = total + _AFT.monitored_events[name].receivedCount
-				matched = name
-			end
-		end
-		if matched then
-			eventGroupCpy[matched] = nil
-			matched = nil
-		end
-		if table_size(eventGroupCpy) == 0 then return total end
+	if err then
+		return 0
+	end
+
+	for event in pairs(eventGroup) do
+		count = count + _AFT.monitored_events[event].receivedCount
 	end
-	return 0
+
+	return count
 end
 
 --[[
@@ -232,51 +231,53 @@ end
 ]]
 
 function _AFT.assertEvtGrpNotReceived(eventGroup, timeout)
-		local count = 0
-		local expected = 0
+		local totalCount = 0
+		local totalExpected = 0
 		local eventName = ""
 
+		for event,expectedCount in pairs(eventGroup) do
+			eventName = eventName .. " " .. event
+			totalExpected = totalExpected + expectedCount
+		end
+
 		if timeout then
-			count = _AFT.lockWaitGroup(eventGroup, timeout)
+			totalCount = _AFT.lockWaitGroup(eventGroup, timeout)
 		else
 			for event in pairs(eventGroup) do
-				count = count + _AFT.monitored_events[event].receivedCount
+				totalCount = totalCount + _AFT.monitored_events[event].receivedCount
 			end
 		end
 
-		for event,expectedCount in pairs(eventGroup) do
-			eventName = eventName .. " " .. event
-			expected = expected + expectedCount
-		end
-		_AFT.assertIsTrue(count <= expected, "One of the following events has been received: '".. eventName .."' but it shouldn't")
+		_AFT.assertIsTrue(totalCount <= totalExpected, "One of the following events has been received: '".. eventName .."' but it shouldn't")
 
 		for event in pairs(eventGroup) do
 			_AFT.triggerEvtCallback(event)
+			_AFT.monitored_events[event] = nil
 		end
 	end
 
 function _AFT.assertEvtGrpReceived(eventGroup, timeout)
-	local count = 0
-	local expected = 0
+	local totalCount = 0
+	local totalExpected = 0
 	local eventName = ""
 
+	for event,expectedCount in pairs(eventGroup) do
+		eventName = eventName .. " " .. event
+		totalExpected = totalExpected + expectedCount
+	end
+
 	if timeout then
-		count = _AFT.lockWaitGroup(eventGroup, timeout)
+		totalCount = _AFT.lockWaitGroup(eventGroup, timeout)
 	else
 		for event in pairs(eventGroup) do
-			count = count + _AFT.monitored_events[event].receivedCount
+			totalCount = totalCount + _AFT.monitored_events[event].receivedCount
 		end
 	end
-
-	for event,expectedCount in pairs(eventGroup) do
-		eventName = eventName .. " " .. event
-		expected = expected + expectedCount
-	end
-
-	_AFT.assertIsTrue(count >= expected, "None or one of the following events: '".. eventName .."' has not been received")
+	_AFT.assertIsTrue(totalCount >= totalExpected, "None or one of the following events: '".. eventName .."' has not been received")
 
 	for event in pairs(eventGroup) do
 		_AFT.triggerEvtCallback(event)
+		_AFT.monitored_events[event] = nil
 	end
 end
 
@@ -289,6 +290,7 @@ function _AFT.assertEvtNotReceived(eventName, timeout)
 		_AFT.assertIsTrue(count == 0, "Event '".. eventName .."' received but it shouldn't")
 
 		_AFT.triggerEvtCallback(eventName)
+		_AFT.monitored_events[eventName] = nil
 	end
 
 function _AFT.assertEvtReceived(eventName, timeout)
@@ -300,6 +302,7 @@ function _AFT.assertEvtReceived(eventName, timeout)
 	_AFT.assertIsTrue(count > 0, "No event '".. eventName .."' received")
 
 	_AFT.triggerEvtCallback(eventName)
+	_AFT.monitored_events[eventName] = nil
 end
 
 function _AFT.testEvtNotReceived(testName, eventName, timeout, setUp, tearDown)
@@ -658,13 +661,13 @@ end
 function _launch_test(context, confArgs, queryArgs)
 	_AFT.context = context
 	_AFT.bindingRootDir = AFB:getrootdir(_AFT.context)
+	_AFT.apiname = AFB:getapiname(_AFT.context)
 
 	-- Enable the lava additionals output markers
 	if queryArgs and queryArgs.lavaOutput then _AFT.lavaOutput = queryArgs.lavaOutput end
 
 	-- Prepare the tests execution configuring the monitoring and loading
 	-- lua test files to execute in the Framework.
-	AFB:servsync(_AFT.context, "monitor", "set", { verbosity = "debug" })
 	if type(confArgs.trace) == "string" then
 		AFB:servsync(_AFT.context, "monitor", "trace", { add = {event = "push_after", pattern = confArgs.trace.."/*" }})
 	elseif type(confArgs.trace) == "table" then
diff --git a/src/aft.c b/src/aft.c
index 01bfaa2..9500368 100644
--- a/src/aft.c
+++ b/src/aft.c
@@ -18,8 +18,10 @@
 
 #define _GNU_SOURCE
 #include <stdio.h>
-#include <string.h>
 #include <time.h>
+#include <pthread.h>
+#include <string.h>
+#include <systemd/sd-event.h>
 
 #include "aft.h"
 #include "mapis.h"
@@ -29,6 +31,9 @@
 static CtlConfigT *CtrlLoadConfigJson(afb_api_t apiHandle, json_object *configJ);
 static CtlConfigT *CtrlLoadConfigFile(afb_api_t apiHandle, const char *configPath);
 static int CtrlCreateApi(afb_api_t apiHandle, CtlConfigT *ctrlConfig);
+static pthread_mutex_t memo_lock;
+static afb_req_t memo_sync = NULL;
+static struct sd_event_source *timersrc = NULL;
 
 // Config Section definition
 static CtlSectionT ctrlSections[] = {
@@ -79,15 +84,67 @@ static void ctrlapi_load(afb_req_t request) {
 
 static void ctrlapi_exit(afb_req_t request) {
 	AFB_REQ_NOTICE(request, "Exiting...");
+	pthread_mutex_destroy(&memo_lock);
 	afb_req_success(request, NULL, NULL);
 	exit(0);
 }
 
+static int timeoutCB(struct sd_event_source *s, uint64_t us, void *ud)
+{
+	if (memo_sync)
+		afb_req_reply(memo_sync, NULL, "timeout", NULL);
+	memo_sync = NULL;
+	timersrc = NULL;
+
+	return 0;
+}
+/**
+ * @brief A verb to call synchronously that will end when a timeout expires or
+ * when a call with a 'stop' order given in the arguments.
+ *
+ * @param request: the AFB request object
+ */
+static void ctrlapi_sync(afb_req_t request) {
+	struct json_object *obj, *val;
+	uint64_t to, usec;
+
+	AFB_REQ_NOTICE(request, "Syncing...");
+
+	pthread_mutex_lock(&memo_lock);
+	obj = afb_req_json(request);
+	if (json_object_object_get_ex(obj, "start", &val)) {
+		to = json_object_get_int(val);
+		if (memo_sync)
+			afb_req_reply(request, NULL, "Bad-State", "There is an already ongoing waiting request.");
+		else {
+			sd_event_now(afb_api_get_event_loop(afb_req_get_api(request)), CLOCK_MONOTONIC, &usec);
+			usec = to + usec;
+			sd_event_add_time(afb_api_get_event_loop(afb_req_get_api(request)), &timersrc, CLOCK_MONOTONIC, usec, 0, timeoutCB, NULL);
+			memo_sync = afb_req_addref(request);
+		}
+	} else if (json_object_object_get_ex(obj, "stop", &val)) {
+		if (!memo_sync)
+			afb_req_reply(request, NULL, "Bad-State", "There isn't any ongoing waiting request.");
+		else {
+			afb_req_reply(memo_sync, json_object_get(val), NULL, NULL);
+			afb_req_unref(memo_sync);
+			sd_event_source_unref(timersrc);
+			memo_sync = NULL;
+			timersrc = NULL;
+			afb_req_reply(request, NULL, NULL, NULL);
+		}
+	} else
+		afb_req_reply(request, NULL, "Invalid", "No 'start' nor 'stop' order provided.");
+
+	pthread_mutex_unlock(&memo_lock);
+}
+
 static afb_verb_t CtrlApiVerbs[] = {
 	/* VERB'S NAME         FUNCTION TO CALL         SHORT DESCRIPTION */
 	{.verb = "ping", .callback = ctrlapi_ping, .info = "ping test for API"},
 	{.verb = "load", .callback = ctrlapi_load, .info = "load a API meant to launch test for a binding"},
 	{.verb = "exit", .callback = ctrlapi_exit, .info = "Exit test"},
+	{.verb = "sync", .callback = ctrlapi_sync, .info = "Manually make a sync for something using a synchronous subcall"},
 	{.verb = NULL} /* marker for end of the array */
 };
 
@@ -112,6 +169,11 @@ static int CtrlInitOneApi(afb_api_t apiHandle) {
 static int CtrlLoadOneApi(void *cbdata, afb_api_t apiHandle) {
 	CtlConfigT *ctrlConfig = (CtlConfigT *)cbdata;
 
+	if(pthread_mutex_init(&memo_lock, NULL)) {
+		AFB_API_ERROR(apiHandle, "Fail to initialize");
+		return -1;
+	}
+
 	// save closure as api's data context
 	afb_api_set_userdata(apiHandle, ctrlConfig);
 
@@ -178,7 +240,7 @@ static int CtrlCreateApi(afb_api_t apiHandle, CtlConfigT *ctrlConfig) {
 	wrap_json_object_add(ctrlConfig->configJ, resourcesJ);
 	wrap_json_object_add(ctrlConfig->configJ, eventsJ);
 
-	if(! afb_api_new_api(apiHandle, ctrlConfig->api, ctrlConfig->info, 1, CtrlLoadOneApi, ctrlConfig))
+	if(! afb_api_new_api(apiHandle, ctrlConfig->api, ctrlConfig->info, 0, CtrlLoadOneApi, ctrlConfig))
 		return ERROR;
 
 	return 0;