3 files changed, 246 insertions, 27 deletions

diff --git a/plugins/afm-main-plugin/afm-main-plugin.c b/plugins/afm-main-plugin/afm-main-plugin.c
index bca28fe0..12bd9f33 100644
--- a/plugins/afm-main-plugin/afm-main-plugin.c
+++ b/plugins/afm-main-plugin/afm-main-plugin.c
@@ -21,17 +21,60 @@
 
 #include "utils-jbus.h"
 
-static const char _id_[] = "id";
+static const char _id_[]        = "id";
+static const char _runid_[]     = "runid";
+static char _runnables_[] = "runnables";
+static char _detail_[]    = "detail";
+static char _start_[]     = "start";
+static char _terminate_[] = "terminate";
+static char _stop_[]      = "stop";
+static char _continue_[]  = "continue";
+static char _runners_[]   = "runners";
+static char _state_[]     = "state";
+static char _install_[]   = "install";
+static char _uninstall_[] = "uninstall";
+
 static struct jbus *jbus;
 
-static struct json_object *call_void(AFB_request *request)
+static struct json_object *embed(AFB_request *request, const char *tag, struct json_object *obj)
+{
+	struct json_object *result;
+
+	if (obj == NULL)
+		result = NULL;
+	else if (!tag) {
+		request->errcode = MHD_HTTP_OK;
+		result = obj;
+	}
+	else {
+		result = json_object_new_object();
+		if (result == NULL) {
+			/* can't embed */
+			result = obj;
+			request->errcode = MHD_HTTP_INTERNAL_SERVER_ERROR;
+		}
+		else {
+			/* TODO why is json-c not returning a status? */
+			json_object_object_add(result, tag, obj);
+			request->errcode = MHD_HTTP_OK;
+		}
+	}
+	return result;
+}
+
+static struct json_object *call(AFB_request *request, AFB_PostItem *item, const char *tag, struct json_object *(*fun)(AFB_request*,AFB_PostItem*))
+{
+	return embed(request, tag, fun(request, item));
+}
+
+static struct json_object *call_void(AFB_request *request, AFB_PostItem *item)
 {
 	struct json_object *obj = jbus_call_sj_sync(jbus, request->api, "true");
 	request->errcode = obj ? MHD_HTTP_OK : MHD_HTTP_FAILED_DEPENDENCY;
 	return obj;
 }
 
-static struct json_object *call_appid(AFB_request *request)
+static struct json_object *call_appid(AFB_request *request, AFB_PostItem *item)
 {
 	struct json_object *obj;
 	char *sid;
@@ -50,10 +93,10 @@ static struct json_object *call_appid(AFB_request *request)
 	return obj;
 }
 
-static struct json_object *call_runid(AFB_request *request)
+static struct json_object *call_runid(AFB_request *request, AFB_PostItem *item)
 {
 	struct json_object *obj;
-	const char *id = getQueryValue(request, _id_);
+	const char *id = getQueryValue(request, _runid_);
 	if (id == NULL) {
 		request->errcode = MHD_HTTP_BAD_REQUEST;
 		return NULL;
@@ -63,16 +106,55 @@ static struct json_object *call_runid(AFB_request *request)
 	return obj;
 }
 
+static struct json_object *call_void__runnables(AFB_request *request, AFB_PostItem *item)
+{
+	return embed(request, _runnables_, call_void(request, item));
+}
+
+static struct json_object *call_appid__runid(AFB_request *request, AFB_PostItem *item)
+{
+	return embed(request, _runid_, call_appid(request, item));
+}
+
+static struct json_object *call_void__runners(AFB_request *request, AFB_PostItem *item)
+{
+	return embed(request, _runners_, call_void(request, item));
+}
+
+static struct json_object *call_file__appid(AFB_request *request, AFB_PostItem *item)
+{
+	if (item == NULL) {
+		struct json_object *obj;
+		char *query;
+		const char *filename = getPostPath(request);
+		request->jresp = NULL;
+		if (0 >= asprintf(&query, "\"%s\"", filename))
+			request->errcode = MHD_HTTP_INTERNAL_SERVER_ERROR;
+		else {
+			obj = jbus_call_sj_sync(jbus, request->api, query);
+			free(query);
+			if (obj)
+				request->jresp = embed(request, _id_, obj);
+			else
+				request->errcode = MHD_HTTP_FAILED_DEPENDENCY;
+		}
+		unlink(filename);
+	}
+	return getPostFile (request, item, "/tmp");
+}
+
 static AFB_restapi plug_apis[] =
 {
-	{"runnables", AFB_SESSION_CHECK, (AFB_apiCB)call_void,  "Get list of runnable applications"},
-	{"detail"   , AFB_SESSION_CHECK, (AFB_apiCB)call_appid, "Get the details for one application"},
-	{"start"    , AFB_SESSION_CHECK, (AFB_apiCB)call_appid, "Start an application"},
-	{"terminate", AFB_SESSION_CHECK, (AFB_apiCB)call_runid, "Terminate a running application"},
-	{"stop"     , AFB_SESSION_CHECK, (AFB_apiCB)call_runid, "Stop (pause) a running application"},
-	{"continue" , AFB_SESSION_CHECK, (AFB_apiCB)call_runid, "Continue (resume) a stopped application"},
-	{"runners"  , AFB_SESSION_CHECK, (AFB_apiCB)call_void,  "Get the list of running applications"},
-	{"state"    , AFB_SESSION_CHECK, (AFB_apiCB)call_runid, "Get the state of a running application"},
+	{_runnables_, AFB_SESSION_CHECK, (AFB_apiCB)call_void__runnables,  "Get list of runnable applications"},
+	{_detail_   , AFB_SESSION_CHECK, (AFB_apiCB)call_appid, "Get the details for one application"},
+	{_start_    , AFB_SESSION_CHECK, (AFB_apiCB)call_appid__runid, "Start an application"},
+	{_terminate_, AFB_SESSION_CHECK, (AFB_apiCB)call_runid, "Terminate a running application"},
+	{_stop_     , AFB_SESSION_CHECK, (AFB_apiCB)call_runid, "Stop (pause) a running application"},
+	{_continue_ , AFB_SESSION_CHECK, (AFB_apiCB)call_runid, "Continue (resume) a stopped application"},
+	{_runners_  , AFB_SESSION_CHECK, (AFB_apiCB)call_void__runners,  "Get the list of running applications"},
+	{_state_    , AFB_SESSION_CHECK, (AFB_apiCB)call_runid, "Get the state of a running application"},
+	{_install_  , AFB_SESSION_CHECK, (AFB_apiCB)call_file__appid,  "Install an application using a widget file"},
+	{_uninstall_, AFB_SESSION_CHECK, (AFB_apiCB)call_appid, "Uninstall an application"},
 	{NULL}
 };
 
diff --git a/plugins/afm-main-plugin/utils-jbus.c b/plugins/afm-main-plugin/utils-jbus.c
index 201b0a6f..e379921c 100644
--- a/plugins/afm-main-plugin/utils-jbus.c
+++ b/plugins/afm-main-plugin/utils-jbus.c
@@ -21,6 +21,8 @@
 #include <stdio.h>
 #include <errno.h>
 #include <string.h>
+#include <poll.h>
+#include <assert.h>
 
 #include <json.h>
 #include <dbus/dbus.h>
@@ -77,6 +79,9 @@ struct jbus {
 	struct jrespw *waiters;
 	char *path;
 	char *name;
+	int watchnr;
+	int watchfd;
+	int watchflags;
 };
 
 /*********************** STATIC COMMON METHODS *****************/
@@ -156,7 +161,11 @@ static int add_signal(
 
 	/* record the signal */
 	if (jbus->signals == NULL) {
+#if 0
+		if (0 >= asprintf(&rule, "type='signal',interface='%s',path='%s'", jbus->name, jbus->path))
+#else
 		if (0 >= asprintf(&rule, "type='signal',sender='%s',interface='%s',path='%s'", jbus->name, jbus->name, jbus->path))
+#endif
 			return -1;
 		dbus_bus_add_match(jbus->connection, rule, NULL);
 		free(rule);
@@ -376,6 +385,61 @@ static DBusHandlerResult incoming(DBusConnection *connection, DBusMessage *messa
 	return DBUS_HANDLER_RESULT_NOT_YET_HANDLED;
 }
 
+static void watchset(DBusWatch *watch, struct jbus *jbus)
+{
+	unsigned int flags;
+	int wf, e;
+
+	flags = dbus_watch_get_flags(watch);
+	e = dbus_watch_get_enabled(watch);
+	wf = jbus->watchflags;
+	if (e) {
+		if (flags & DBUS_WATCH_READABLE)
+			wf |= POLLIN;
+		if (flags & DBUS_WATCH_WRITABLE)
+			wf |= POLLOUT;
+	}
+	else {
+		if (flags & DBUS_WATCH_READABLE)
+			wf &= ~POLLIN;
+		if (flags & DBUS_WATCH_WRITABLE)
+			wf &= ~POLLOUT;
+	}
+	jbus->watchflags = wf;
+}
+
+static void watchdel(DBusWatch *watch, void *data)
+{
+	struct jbus *jbus = data;
+
+	assert(jbus->watchnr > 0);
+	assert(jbus->watchfd == dbus_watch_get_unix_fd(watch));
+	jbus->watchnr--;
+}
+
+static void watchtoggle(DBusWatch *watch, void *data)
+{
+	struct jbus *jbus = data;
+
+	assert(jbus->watchnr > 0);
+	assert(jbus->watchfd == dbus_watch_get_unix_fd(watch));
+	watchset(watch, jbus);
+}
+
+static dbus_bool_t watchadd(DBusWatch *watch, void *data)
+{
+	struct jbus *jbus = data;
+	if (jbus->watchnr == 0) {
+		jbus->watchfd = dbus_watch_get_unix_fd(watch);
+		jbus->watchflags = 0;
+	}
+	else if (jbus->watchfd != dbus_watch_get_unix_fd(watch))
+		return FALSE;
+	jbus->watchnr++;
+	watchset(watch, jbus);
+	return TRUE;
+}
+
 /************************** MAIN FUNCTIONS *****************************************/
 
 struct jbus *create_jbus(int session, const char *path)
@@ -416,12 +480,10 @@ struct jbus *create_jbus(int session, const char *path)
 
 	/* connect */
 	jbus->connection = dbus_bus_get(session ? DBUS_BUS_SESSION : DBUS_BUS_SYSTEM, NULL);
-	if (jbus->connection == NULL) {
+	if (jbus->connection == NULL
+	|| !dbus_connection_add_filter(jbus->connection, incoming, jbus, NULL)
+        || !dbus_connection_set_watch_functions(jbus->connection, watchadd, watchdel, watchtoggle, jbus, NULL))
 		goto error2;
-	}
-	if (!dbus_connection_add_filter(jbus->connection, incoming, jbus, NULL)) {
-		goto error2;
-	}
 
 	return jbus;
 
@@ -512,7 +574,8 @@ int jbus_send_signal_s(struct jbus *jbus, const char *name, const char *content)
 	if (message == NULL)
 		goto error;
 
-	if (!dbus_message_append_args(message, DBUS_TYPE_STRING, &content, DBUS_TYPE_INVALID)) {
+	if (!dbus_message_set_sender(message, jbus->name)
+	||  !dbus_message_append_args(message, DBUS_TYPE_STRING, &content, DBUS_TYPE_INVALID)) {
 		dbus_message_unref(message);
 		goto error;
 	}
@@ -562,12 +625,81 @@ int jbus_start_serving(struct jbus *jbus)
 	}
 }
 
+int jbus_fill_pollfds(struct jbus **jbuses, int njbuses, struct pollfd *fds)
+{
+	int i, r;
+
+	for (r = i = 0 ; i < njbuses ; i++) {
+		if (jbuses[i]->watchnr) {
+			fds[r].fd = jbuses[i]->watchfd;
+			fds[r].events = jbuses[i]->watchflags;
+			r++;
+		}
+	}
+	return r;
+}
+
+int jbus_dispatch_pollfds(struct jbus **jbuses, int njbuses, struct pollfd *fds, int maxcount)
+{
+	int i, r, n;
+	DBusDispatchStatus sts;
+
+	for (r = n = i = 0 ; i < njbuses && n < maxcount ; i++) {
+		if (jbuses[i]->watchnr && fds[r].fd == jbuses[i]->watchfd) {
+			if (fds[r].revents) {
+				dbus_connection_read_write(jbuses[i]->connection, 0);
+				sts = dbus_connection_get_dispatch_status(jbuses[i]->connection);
+				while(sts == DBUS_DISPATCH_DATA_REMAINS &&  n < maxcount) {
+					sts = dbus_connection_dispatch(jbuses[i]->connection);
+					n++;
+				}
+			}
+			r++;
+		}
+	}
+	return n;
+}
+
+int jbus_dispatch_multiple(struct jbus **jbuses, int njbuses, int maxcount)
+{
+	int i, r;
+	DBusDispatchStatus sts;
+
+	for (i = r = 0 ; i < njbuses && r < maxcount ; i++) {
+		dbus_connection_read_write(jbuses[i]->connection, 0);
+		while(sts == DBUS_DISPATCH_DATA_REMAINS &&  r < maxcount) {
+			sts = dbus_connection_dispatch(jbuses[i]->connection);
+			r++;
+		}
+	}
+	return r;
+}
+
+int jbus_read_write_dispatch_multiple(struct jbus **jbuses, int njbuses, int toms, int maxcount)
+{
+	int n, r, s;
+	struct pollfd *fds;
+
+	if (njbuses < 0 || njbuses > 100) {
+		errno = EINVAL;
+		return -1;
+	}
+	fds = alloca(njbuses * sizeof * fds);
+	assert(fds != NULL);
+
+	r = jbus_dispatch_multiple(jbuses, njbuses, maxcount);
+	n = jbus_fill_pollfds(jbuses, njbuses, fds);
+	s = poll(fds, n, toms);
+	if (s < 0)
+		return r ? r : s;
+	n = jbus_dispatch_pollfds(jbuses, njbuses, fds, maxcount - r);
+	return n >= 0 ? r + n : r ? r : n;
+}
+
 int jbus_read_write_dispatch(struct jbus *jbus, int toms)
 {
-	if (dbus_connection_read_write_dispatch(jbus->connection, toms));
-		return 0;
-	errno = EPIPE;
-	return -1;
+	int r = jbus_read_write_dispatch_multiple(&jbus, 1, toms, 1000);
+	return r < 0 ? r : 0;
 }
 
 int jbus_call_ss(struct jbus *jbus, const char *method, const char *query, void (*onresp)(int, const char*, void*), void *data)
@@ -682,8 +814,7 @@ int main()
 	s2 = jbus_add_service_j(jbus, "incr", incr);
 	s3 = jbus_start_serving(jbus);
 	printf("started %d %d %d\n", s1, s2, s3);
-	while (!jbus_read_write_dispatch (jbus, -1))
-		;
+	while (!jbus_read_write_dispatch (jbus, -1));
 }
 #endif
 #ifdef CLIENT
@@ -710,8 +841,7 @@ int main()
 		jbus_read_write_dispatch (jbus, 1);
 	}
 	printf("[[[%s]]]\n", jbus_call_ss_sync(jbus, "ping", "\"formidable!\""));
-	while (!jbus_read_write_dispatch (jbus, -1))
-		;
+	while (!jbus_read_write_dispatch (jbus, -1));
 }
 #endif
 
diff --git a/plugins/afm-main-plugin/utils-jbus.h b/plugins/afm-main-plugin/utils-jbus.h
index 218c21d6..3b6ec75f 100644
--- a/plugins/afm-main-plugin/utils-jbus.h
+++ b/plugins/afm-main-plugin/utils-jbus.h
@@ -24,6 +24,12 @@ extern struct jbus *create_jbus(int session, const char *path);
 extern void jbus_addref(struct jbus *jbus);
 extern void jbus_unref(struct jbus *jbus);
 
+struct pollfd;
+extern int jbus_fill_pollfds(struct jbus **jbuses, int njbuses, struct pollfd *fds);
+extern int jbus_dispatch_pollfds(struct jbus **jbuses, int njbuses, struct pollfd *fds, int maxcount);
+extern int jbus_read_write_dispatch_multiple(struct jbus **jbuses, int njbuses, int toms, int maxcount);
+extern int jbus_dispatch_multiple(struct jbus **jbuses, int njbuses, int maxcount);
+
 extern int jbus_read_write_dispatch(struct jbus *jbus, int toms);
 
 /* verbs for the clients */
@@ -52,3 +58,4 @@ extern int jbus_start_serving(struct jbus *jbus);
 extern int jbus_send_signal_s(struct jbus *jbus, const char *name, const char *content);
 extern int jbus_send_signal_j(struct jbus *jbus, const char *name, struct json_object *content);
 
+