From 7fc272e5a9afc922f14c47b4c27d0a564e2de4ca Mon Sep 17 00:00:00 2001 From: =?utf8?q?Jos=C3=A9=20Bollo?= Date: Mon, 25 Jan 2016 23:47:05 +0100 Subject: [PATCH] utils-jbus: implments multiple bus handling MIME-Version: 1.0 Content-Type: text/plain; charset=utf8 Content-Transfer-Encoding: 8bit Change-Id: I3c40b27448000fda800e2a3d98210eb6c11bb969 Signed-off-by: José Bollo --- src/afm-user-daemon.c | 35 ++++++----- src/utils-jbus.c | 158 +++++++++++++++++++++++++++++++++++++++++++++----- src/utils-jbus.h | 7 +++ 3 files changed, 168 insertions(+), 32 deletions(-) diff --git a/src/afm-user-daemon.c b/src/afm-user-daemon.c index 87d7e71..f6d2aef 100644 --- a/src/afm-user-daemon.c +++ b/src/afm-user-daemon.c @@ -56,8 +56,7 @@ static struct option options[] = { { NULL, 0, NULL, 0 } }; -static struct jbus *jbus; -static struct jbus *jbusys; +static struct jbus *jbuses[2]; static struct afm_db *afdb; const char error_nothing[] = "[]"; @@ -174,7 +173,7 @@ static void on_signal_changed(struct json_object *obj) /* update the database */ afm_db_update_applications(afdb); /* propagate now */ - jbus_send_signal_j(jbus, "changed", obj); + jbus_send_signal_j(jbuses[1], "changed", obj); } static int daemonize() @@ -273,40 +272,40 @@ int main(int ac, char **av) } /* init observers */ - jbusys = create_jbus(0, AFM_SYSTEM_DBUS_PATH); - if (!jbusys) { + jbuses[0] = create_jbus(0, AFM_SYSTEM_DBUS_PATH); + if (!jbuses[0]) { ERROR("create_jbus failed for system"); return 1; } - if(jbus_on_signal_j(jbusys, "changed", on_signal_changed)) { + if(jbus_on_signal_j(jbuses[0], "changed", on_signal_changed)) { ERROR("adding signal observer failed"); return 1; } /* init service */ - jbus = create_jbus(1, AFM_USER_DBUS_PATH); - if (!jbus) { + jbuses[1] = create_jbus(1, AFM_USER_DBUS_PATH); + if (!jbuses[1]) { ERROR("create_jbus failed"); return 1; } - if(jbus_add_service_j(jbus, "runnables", on_runnables) - || jbus_add_service_j(jbus, "detail", on_detail) - || jbus_add_service_j(jbus, "start", on_start) - || jbus_add_service_j(jbus, "terminate", on_terminate) - || jbus_add_service_j(jbus, "stop", on_stop) - || jbus_add_service_j(jbus, "continue", on_continue) - || jbus_add_service_j(jbus, "runners", on_runners) - || jbus_add_service_j(jbus, "state", on_state)) { + if(jbus_add_service_j(jbuses[1], "runnables", on_runnables) + || jbus_add_service_j(jbuses[1], "detail", on_detail) + || jbus_add_service_j(jbuses[1], "start", on_start) + || jbus_add_service_j(jbuses[1], "terminate", on_terminate) + || jbus_add_service_j(jbuses[1], "stop", on_stop) + || jbus_add_service_j(jbuses[1], "continue", on_continue) + || jbus_add_service_j(jbuses[1], "runners", on_runners) + || jbus_add_service_j(jbuses[1], "state", on_state)) { ERROR("adding services failed"); return 1; } /* start and run */ - if (jbus_start_serving(jbus)) { + if (jbus_start_serving(jbuses[1])) { ERROR("can't start server"); return 1; } - while (!jbus_read_write_dispatch(jbus, -1)); + while (jbus_read_write_dispatch_multiple(jbuses, 2, -1, 20) >= 0); return 0; } diff --git a/src/utils-jbus.c b/src/utils-jbus.c index 201b0a6..e379921 100644 --- a/src/utils-jbus.c +++ b/src/utils-jbus.c @@ -21,6 +21,8 @@ #include #include #include +#include +#include #include #include @@ -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/src/utils-jbus.h b/src/utils-jbus.h index 218c21d..3b6ec75 100644 --- a/src/utils-jbus.h +++ b/src/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); + -- 2.16.6