initial event handler

Change-Id: Idb92d6de9904d050b37ef0a5d664e82573ff640d
Signed-off-by: José Bollo <jose.bollo@iot.bzh>

7 files changed, 199 insertions, 12 deletions

diff --git a/plugins/samples/HelloWorld.c b/plugins/samples/HelloWorld.c
index bf809cc3..ec060e97 100644
--- a/plugins/samples/HelloWorld.c
+++ b/plugins/samples/HelloWorld.c
@@ -21,13 +21,14 @@
 
 #include "afb-plugin.h"
 
+const struct AFB_interface *interface;
+
 // Sample Generic Ping Debug API
 static void ping(struct afb_req request, json_object *jresp, const char *tag)
 {
-    static int pingcount = 0;
-    json_object *query = afb_req_json(request);
-
-    afb_req_success_f(request, jresp, "Ping Binder Daemon tag=%s count=%d query=%s", tag, ++pingcount, json_object_to_json_string(query));
+	static int pingcount = 0;
+	json_object *query = afb_req_json(request);
+	afb_req_success_f(request, jresp, "Ping Binder Daemon tag=%s count=%d query=%s", tag, ++pingcount, json_object_to_json_string(query));
 }
 
 static void pingSample (struct afb_req request)
@@ -50,6 +51,13 @@ static void pingBug (struct afb_req request)
 	ping((struct afb_req){NULL,NULL,NULL}, NULL, "pingBug");
 }
 
+static void pingEvent(struct afb_req request)
+{
+	json_object *query = afb_req_json(request);
+	afb_evmgr_push(afb_daemon_get_evmgr(interface->daemon), "event", json_object_get(query));
+	ping(request, json_object_get(query), "event");
+}
+
 
 // For samples https://linuxprograms.wordpress.com/2010/05/20/json-c-libjson-tutorial/
 static void pingJson (struct afb_req request) {
@@ -76,6 +84,7 @@ static const struct AFB_restapi pluginApis[]= {
   {"pingnull" , AFB_SESSION_NONE, pingNull    , "Return NULL"},
   {"pingbug"  , AFB_SESSION_NONE, pingBug     , "Do a Memory Violation"},
   {"pingJson" , AFB_SESSION_NONE, pingJson    , "Return a JSON object"},
+  {"pingevent", AFB_SESSION_NONE, pingEvent   , "Send an event"},
   {NULL}
 };
 
@@ -88,5 +97,6 @@ static const struct AFB_plugin plugin_desc = {
 
 const struct AFB_plugin *pluginRegister (const struct AFB_interface *itf)
 {
+	interface = itf;
 	return &plugin_desc;
 }
diff --git a/src/afb-api-so.c b/src/afb-api-so.c
index c3cb08d7..88246b58 100644
--- a/src/afb-api-so.c
+++ b/src/afb-api-so.c
@@ -46,6 +46,7 @@ extern __thread sigjmp_buf *error_handler;
 
 struct api_so_desc {
 	struct AFB_plugin *plugin;	/* descriptor */
+	size_t apilength;
 	void *handle;			/* context of dlopen */
 	struct AFB_interface interface;	/* interface */
 };
@@ -65,6 +66,16 @@ static const struct afb_pollmgr_itf pollmgr_itf = {
 
 static void afb_api_so_evmgr_push(struct api_so_desc *desc, const char *name, struct json_object *object)
 {
+	size_t length;
+	char *event;
+
+	assert(desc->plugin != NULL);
+	length = strlen(name);
+	event = alloca(length + 2 + desc->apilength);
+	memcpy(event, desc->plugin->prefix, desc->apilength);
+	event[desc->apilength] = '/';
+	memcpy(event + desc->apilength + 1, name, length + 1);
+	ctxClientEventSend(NULL, event, object);
 }
 
 static const struct afb_evmgr_itf evmgr_itf = {
@@ -224,6 +235,7 @@ int afb_api_so_add_plugin(const char *path)
 	}
 
 	/* records the plugin */
+	desc->apilength = strlen(desc->plugin->prefix);
 	if (afb_apis_add(desc->plugin->prefix, (struct afb_api){
 			.closure = desc,
 			.call = (void*)call}) < 0) {
diff --git a/src/afb-ws-json.c b/src/afb-ws-json.c
index a34142c9..471831e7 100644
--- a/src/afb-ws-json.c
+++ b/src/afb-ws-json.c
@@ -52,7 +52,14 @@ struct afb_ws_json
 	struct afb_ws *ws;
 };
 
-struct afb_ws_json *afb_ws_json_create(int fd, struct AFB_clientCtx *context, void (*cleanup)(void*), void *closure)
+
+static void aws_send_event(struct afb_ws_json *ws, const char *event, struct json_object *object);
+
+static const struct afb_event_sender_itf event_sender_itf = {
+	.send = (void*)aws_send_event
+};
+
+struct afb_ws_json *afb_ws_json_create(int fd, struct AFB_clientCtx *context, void (*cleanup)(void*), void *cleanup_closure)
 {
 	struct afb_ws_json *result;
 
@@ -64,7 +71,7 @@ struct afb_ws_json *afb_ws_json_create(int fd, struct AFB_clientCtx *context, vo
 		goto error;
 
 	result->cleanup = cleanup;
-	result->cleanup_closure = closure;
+	result->cleanup_closure = cleanup_closure;
 	result->requests = NULL;
 	result->context = ctxClientGet(context);
 	if (result->context == NULL)
@@ -78,8 +85,13 @@ struct afb_ws_json *afb_ws_json_create(int fd, struct AFB_clientCtx *context, vo
 	if (result->ws == NULL)
 		goto error4;
 
+	if (0 > ctxClientEventSenderAdd(result->context, (struct afb_event_sender){ .itf = &event_sender_itf, .closure = result }))
+		goto error5;
+
 	return result;
 
+error5:
+	/* TODO */
 error4:
 	json_tokener_free(result->tokener);
 error3:
@@ -100,6 +112,7 @@ static void aws_on_close(struct afb_ws_json *ws, uint16_t code, char *text, size
 #define CALL 2
 #define RETOK 3
 #define RETERR 4
+#define EVENT 5
 
 struct afb_wsreq
 {
@@ -320,8 +333,8 @@ static struct json_object *wsreq_json(struct afb_wsreq *wsreq)
 		json_tokener_reset(wsreq->aws->tokener);
 		root = json_tokener_parse_ex(wsreq->aws->tokener, wsreq->obj, (int)wsreq->objlen);
 		if (root == NULL) {
-			/* lazy discovering !!!! not good TODO improve*/
-			root = json_object_new_object();
+			/* lazy error detection of json request. Is it to improve? */
+			root = json_object_new_string_len(wsreq->obj, (int)wsreq->objlen);
 		}
 		wsreq->root = root;
 	}
@@ -429,3 +442,28 @@ static void wsreq_send(struct afb_wsreq *wsreq, char *buffer, size_t size)
 	afb_ws_text(wsreq->aws->ws, buffer, size);
 }
 
+static void aws_send_event(struct afb_ws_json *aws, const char *event, struct json_object *object)
+{
+	json_object *root, *reply;
+	const char *message;
+
+	/* builds the answering structure */
+	root = json_object_new_object();
+	json_object_object_add(root, "jtype", json_object_new_string("afb-event"));
+	json_object_object_add(root, "event", json_object_new_string(event));
+	if (object)
+		json_object_object_add(root, "data", object);
+
+	/* make the reply */
+	reply = json_object_new_array();
+	json_object_array_add(reply, json_object_new_int(EVENT));
+	json_object_array_add(reply, json_object_new_string(event));
+	json_object_array_add(reply, root);
+	json_object_array_add(reply, json_object_new_string(aws->context->token));
+
+	/* emits the reply */
+	message = json_object_to_json_string(reply);
+	afb_ws_text(aws->ws, message, strlen(message));
+	json_object_put(reply);
+}
+
diff --git a/src/session.c b/src/session.c
index fbc7f2fb..3e08808f 100644
--- a/src/session.c
+++ b/src/session.c
@@ -23,6 +23,7 @@
 #include <string.h>
 #include <uuid/uuid.h>
 #include <assert.h>
+#include <errno.h>
 
 #include "session.h"
 
@@ -113,7 +114,7 @@ static int ctxStoreDel (struct AFB_clientCtx *client)
 
     for (idx=0; idx < sessions.max; idx++) {
         if (sessions.store[idx] == client) {
-	        sessions.store[idx]=NULL;
+	        sessions.store[idx] = NULL;
         	sessions.count--;
 	        status = 1;
 		goto deleted;
@@ -138,7 +139,7 @@ static int ctxStoreAdd (struct AFB_clientCtx *client)
 
     for (idx=0; idx < sessions.max; idx++) {
         if (NULL == sessions.store[idx]) {
-        	sessions.store[idx]= client;
+        	sessions.store[idx] = client;
 	        sessions.count++;
         	status = 1;
 		goto added;
@@ -289,3 +290,95 @@ void ctxTokenNew (struct AFB_clientCtx *clientCtx)
 	clientCtx->expiration = NOW + sessions.timeout;
 }
 
+struct afb_event_sender_list
+{
+	struct afb_event_sender_list *next;
+	struct afb_event_sender sender;
+	int refcount;
+};
+
+int ctxClientEventSenderAdd(struct AFB_clientCtx *clientCtx, struct afb_event_sender sender)
+{
+	struct afb_event_sender_list *iter, **prv;
+
+	prv = &clientCtx->senders;
+	for (;;) {
+		iter = *prv;
+		if (iter == NULL) {
+			iter = calloc(1, sizeof *iter);
+			if (iter == NULL) {
+				errno = ENOMEM;
+				return -1;
+			}
+			iter->sender = sender;
+			iter->refcount = 1;
+			*prv = iter;
+			return 0;
+		}
+		if (iter->sender.itf == sender.itf && iter->sender.closure == sender.closure) {
+			iter->refcount++;
+			return 0;
+		}
+		prv = &iter->next;
+	}
+}
+
+void ctxClientEventSenderRemove(struct AFB_clientCtx *clientCtx, struct afb_event_sender sender)
+{
+	struct afb_event_sender_list *iter, **prv;
+
+	prv = &clientCtx->senders;
+	for (;;) {
+		iter = *prv;
+		if (iter == NULL)
+			return;
+		if (iter->sender.itf == sender.itf && iter->sender.closure == sender.closure) {
+			if (!--iter->refcount) {
+				*prv = iter->next;
+				free(iter);
+			}
+			return;
+		}
+		prv = &iter->next;
+	}
+}
+
+static int send(struct AFB_clientCtx *clientCtx, const char *event, struct json_object *object)
+{
+	struct afb_event_sender_list *iter;
+	int result;
+
+	result = 0;
+	iter = clientCtx->senders;
+	while (iter != NULL) {
+		iter->sender.itf->send(iter->sender.closure, event, object);
+		result++;
+		iter = iter->next;
+	}
+
+	return result;
+}
+
+int ctxClientEventSend(struct AFB_clientCtx *clientCtx, const char *event, struct json_object *object)
+{
+	long idx;
+	time_t now;
+	int result;
+
+	if (clientCtx != NULL)
+		result = send(clientCtx, event, object);
+	else {
+		result = 0;
+		now = NOW;
+		for (idx=0; idx < sessions.max; idx++) {
+			clientCtx = sessions.store[idx];
+			if (clientCtx != NULL && !ctxStoreTooOld(clientCtx, now)) {
+				clientCtx = ctxClientGet(clientCtx);
+				result += send(clientCtx, event, object);
+				ctxClientPut(clientCtx);
+			}
+		}
+	}
+	return result;
+}
+
diff --git a/src/session.h b/src/session.h
index 569af654..7930bdb3 100644
--- a/src/session.h
+++ b/src/session.h
@@ -14,10 +14,11 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-// User Client Session Context
 
 #pragma once
 
+struct json_object;
+
 struct afb_context
 {
 	void *context;
@@ -27,6 +28,19 @@ struct afb_context
 extern void *afb_context_get(struct afb_context *actx);
 extern void afb_context_set(struct afb_context *actx, void *context, void (*free_context)(void*));
 
+struct afb_event_sender_itf
+{
+	void (*send)(void *closure, const char *event, struct json_object *object);
+};
+
+struct afb_event_sender
+{
+	const struct afb_event_sender_itf *itf;
+	void *closure;
+};
+
+struct afb_event_sender_list;
+
 struct AFB_clientCtx
 {
 	time_t expiration;    // expiration time of the token
@@ -35,6 +49,7 @@ struct AFB_clientCtx
 	struct afb_context *contexts;
 	char uuid[37];        // long term authentication of remote client
 	char token[37];       // short term authentication of remote client
+	struct afb_event_sender_list *senders;
 };
 
 extern void ctxStoreInit (int max_session_count, int timeout, const char *initok, int context_count);
@@ -43,6 +58,11 @@ extern struct AFB_clientCtx *ctxClientGetForUuid (const char *uuid);
 extern struct AFB_clientCtx *ctxClientGet(struct AFB_clientCtx *clientCtx);
 extern void ctxClientPut(struct AFB_clientCtx *clientCtx);
 extern void ctxClientClose (struct AFB_clientCtx *clientCtx);
+
+extern int ctxClientEventSenderAdd(struct AFB_clientCtx *clientCtx, struct afb_event_sender sender);
+extern void ctxClientEventSenderRemove(struct AFB_clientCtx *clientCtx, struct afb_event_sender sender);
+extern int ctxClientEventSend(struct AFB_clientCtx *clientCtx, const char *event, struct json_object *object);
+
 extern int ctxTokenCheck (struct AFB_clientCtx *clientCtx, const char *token);
 extern int ctxTokenCheckLen (struct AFB_clientCtx *clientCtx, const char *token, size_t length);
 extern void ctxTokenNew (struct AFB_clientCtx *clientCtx);
diff --git a/test/AFB.js b/test/AFB.js
index 309db472..44b1a908 100644
--- a/test/AFB.js
+++ b/test/AFB.js
@@ -37,12 +37,14 @@ var AFB_websocket;
 	var CALL = 2;
 	var RETOK = 3;
 	var RETERR = 4;
+	var EVENT = 5;
 
 	var PROTO1 = "x-afb-ws-json1";
 
 	AFB_websocket = function(onopen, onabort) {
 		this.ws = new WebSocket(urlws, [ PROTO1 ]);
 		this.pendings = {};
+		this.awaitens = {};
 		this.counter = 0;
 		this.ws.onopen = onopen.bind(this);
 		this.ws.onerror = onerror.bind(this);
@@ -90,6 +92,10 @@ var AFB_websocket;
 			delete this.pendings[id];
 		}
 		switch (code) {
+		case EVENT:
+			var a = this.awaitens[id];
+			if (a)
+				a.forEach(function(handler){handler(ans);});
 		case RETOK:
 			pend && pend.onsuccess && pend.onsuccess(ans, this);
 			break; 
@@ -112,9 +118,16 @@ var AFB_websocket;
 		this.ws.send(JSON.stringify(arr));
 	}
 
+	function onevent(api, name, handler) {
+		var id = api+"/"+name;
+		var list = this.awaitens[id] || (this.awaitens[id] = []);
+		list.push(handler);
+	}
+
 	AFB_websocket.prototype = {
 		close: close,
-		call: call
+		call: call,
+		onevent: onevent
 	};
 }
 /*********************************************/
diff --git a/test/hello-world.html b/test/hello-world.html
index 67464ff5..f8e0f54a 100644
--- a/test/hello-world.html
+++ b/test/hello-world.html
@@ -9,5 +9,6 @@
      <li><a href="api/hello/pingnull">ping null</a>
      <li><a href="api/hello/pingbug">ping bug</a>
      <li><a href="api/hello/pingJson?toto&tata&titi=u">ping json</a>
+     <li><a href="api/hello/pingevent?toto&tata&titi=u">ping event</a>
      <li><a href="api/hello/none">not a verb</a>
      <li><a href="api/none/none">not an api</a>