diff options
author | José Bollo <jose.bollo@iot.bzh> | 2016-04-15 16:40:38 +0200 |
---|---|---|
committer | José Bollo <jose.bollo@iot.bzh> | 2016-04-15 16:40:38 +0200 |
commit | 68a8eaafe5f43480f29308bfd2ec12ad54da43f1 (patch) | |
tree | 5c8782a09cb48df20695fa6ba562e00f523d22f3 /plugins/afm-main-plugin/utils-jbus.c | |
parent | 67966a2f96613d833f493ef3773c442d35f8ed31 (diff) |
use upoll for event loop
also add '#pragma once' in headers
Change-Id: I90cc2d53ec60d4d1e66cf0f229109621e4019864
Signed-off-by: José Bollo <jose.bollo@iot.bzh>
Diffstat (limited to 'plugins/afm-main-plugin/utils-jbus.c')
-rw-r--r-- | plugins/afm-main-plugin/utils-jbus.c | 838 |
1 files changed, 193 insertions, 645 deletions
diff --git a/plugins/afm-main-plugin/utils-jbus.c b/plugins/afm-main-plugin/utils-jbus.c index 3bea2c23..4872987f 100644 --- a/plugins/afm-main-plugin/utils-jbus.c +++ b/plugins/afm-main-plugin/utils-jbus.c @@ -25,8 +25,8 @@ #include <assert.h> #include <json.h> -#include <dbus/dbus.h> +#include "utils-sbus.h" #include "utils-jbus.h" /* @@ -37,21 +37,7 @@ /* * errors messages generated by jbus */ -#if defined(NO_JSON_ERROR_STRING) -static const char invalid_request_string[] = "invalid request"; static const char out_of_memory_string[] = "out of memory"; -#else -static const char invalid_request_string[] = "\"invalid request\""; -static const char out_of_memory_string[] = "\"out of memory\""; -#endif - -/* - * structure for handled requests - */ -struct jreq { - DBusConnection *connection; /* connection of the request */ - DBusMessage *request; /* message of the request */ -}; /* * structure for services @@ -59,9 +45,9 @@ struct jreq { struct jservice { struct jservice *next; /* link to the next service */ char *method; /* method name for the service */ - void (*oncall_s) (struct jreq *, const char *, void *); + void (*oncall_s) (struct sbusmsg *, const char *, void *); /* string callback */ - void (*oncall_j) (struct jreq *, struct json_object *, void *); + void (*oncall_j) (struct sbusmsg *, struct json_object *, void *); /* json callback */ void *data; /* closure data for the callbacks */ }; @@ -80,24 +66,13 @@ struct jsignal { }; /* - * structure for asynchronous requests (resp-onse w-aiter) + * structure for asynchronous requests */ struct jrespw { - struct jrespw *next; /* next asynchronous */ - dbus_uint32_t serial; /* serial dbus number */ - void *data; /* closure data for the callbacks */ - void (*onresp_s) (int, const char *, void *); - /* string callback */ - void (*onresp_j) (int, struct json_object *, void *); + struct jbus *jbus; + void (*onresp) (int, struct json_object *, void *); /* json callback */ -}; - -/* - * structure for synchronous requests - */ -struct respsync { - int replied; /* boolean flag indicating reply */ - char *value; /* copy of the returned value */ + void *data; /* closure data for the callbacks */ }; /* @@ -105,53 +80,71 @@ struct respsync { */ struct jbus { int refcount; /* referenced how many time */ - DBusConnection *connection; /* connection to DBU */ + struct sbus *sbus; + struct sbus_service *sservice; + struct sbus_signal *ssignal; struct json_tokener *tokener; /* string to json tokenizer */ struct jservice *services; /* first service */ struct jsignal *signals; /* first signal */ - struct jrespw *waiters; /* first response waiter */ char *path; /* dbus path */ char *name; /* dbus name */ - int watchnr; /* counter of watching need */ - int watchfd; /* file to watch */ - short watchflags; /* watched flags */ }; /*********************** STATIC COMMON METHODS *****************/ /* - * Frees the ressources attached to a request - */ -static inline void free_jreq(struct jreq *jreq) -{ - dbus_message_unref(jreq->request); - dbus_connection_unref(jreq->connection); - free(jreq); -} - -/* * Replies the error "out of memory". * This function is intended to be used in services when an * allocation fails. Thus, it set errno to ENOMEM and * returns -1. */ -static inline int reply_out_of_memory(struct jreq *jreq) +static inline int reply_out_of_memory(struct sbusmsg *smsg) { - jbus_reply_error_s(jreq, out_of_memory_string); + jbus_reply_error_s(smsg, out_of_memory_string); errno = ENOMEM; return -1; } /* - * Checks if the incoming 'message' matches the interface - * linked to 'jbus'. + * Parses the json-string 'msg' to create a json object stored + * in 'obj'. It uses the tokener of 'jbus'. This is a small + * improvement to avoid recreation of tokeners. * - * Returns 1 if it matches or 0 wether it does not matches. + * Returns 1 in case of success and put the result in *'obj'. + * Returns 0 in case of error and put NULL in *'obj'. */ -static int matchitf(struct jbus *jbus, DBusMessage * message) +static int jparse(struct jbus *jbus, const char *msg, struct json_object **obj) +{ + json_tokener_reset(jbus->tokener); + *obj = json_tokener_parse_ex(jbus->tokener, msg, -1); + if (json_tokener_get_error(jbus->tokener) == json_tokener_success) + return 1; + json_object_put(*obj); + *obj = NULL; + return 0; +} + +static void on_service_call(struct sbusmsg *smsg, const char *content, struct jbus *jbus) { - const char *itf = dbus_message_get_interface(message); - return itf != NULL && !strcmp(itf, jbus->name); + struct jservice *service; + const char *member; + struct json_object *obj; + + member = sbus_member(smsg); + service = jbus->services; + while (service != NULL) { + if (!strcmp(service->method, member)) { + if (service->oncall_s) + service->oncall_s(smsg, content, service->data); + else if (service->oncall_j) { + if (!jparse(jbus, content, &obj)) + obj = json_object_new_string(content); + service->oncall_j(smsg, obj, service->data); + json_object_put(obj); + } + } + service = service->next; + } } /* @@ -165,12 +158,21 @@ static int matchitf(struct jbus *jbus, DBusMessage * message) static int add_service( struct jbus *jbus, const char *method, - void (*oncall_s) (struct jreq *, const char *, void *), - void (*oncall_j) (struct jreq *, struct json_object *, void *), + void (*oncall_s) (struct sbusmsg *, const char *, void *), + void (*oncall_j) (struct sbusmsg *, struct json_object *, void *), void *data) { struct jservice *srv; + /* connection of the service */ + if (jbus->sservice == NULL) { + jbus->sservice = sbus_add_service(jbus->sbus, + NULL, jbus->path, jbus->name, NULL, + (void*)on_service_call, jbus); + if (jbus->sservice == NULL) + goto error; + } + /* allocation */ srv = malloc(sizeof *srv); if (srv == NULL) { @@ -195,9 +197,33 @@ static int add_service( error2: free(srv); error: + errno = ENOMEM; return -1; } +static void on_signal_event(const struct sbusmsg *smsg, const char *content, struct jbus *jbus) +{ + struct jsignal *signal; + const char *member; + struct json_object *obj; + + member = sbus_member(smsg); + signal = jbus->signals; + while (signal != NULL) { + if (!strcmp(signal->name, member)) { + if (signal->onsignal_s) + signal->onsignal_s(content, signal->data); + else if (signal->onsignal_j) { + if (!jparse(jbus, content, &obj)) + obj = json_object_new_string(content); + signal->onsignal_j(obj, signal->data); + json_object_put(obj); + } + } + signal = signal->next; + } +} + /* * Adds to 'jbus' a handler for the signal of 'name' emmited by * the sender and the interface that 'jbus' is linked to. @@ -215,17 +241,15 @@ static int add_signal( void (*onsignal_j) (struct json_object *, void *), void *data) { - char *rule; struct jsignal *sig; - /* record the signal */ - if (jbus->signals == NULL) { - if (0 >= asprintf(&rule, - "type='signal',sender='%s',interface='%s',path='%s'", - jbus->name, jbus->name, jbus->path)) - return -1; - dbus_bus_add_match(jbus->connection, rule, NULL); - free(rule); + /* connection of the signal */ + if (jbus->ssignal == NULL) { + jbus->ssignal = sbus_add_signal(jbus->sbus, + NULL, jbus->path, jbus->name, NULL, + (void*)on_signal_event, jbus); + if (jbus->ssignal == NULL) + goto error; } /* allocation */ @@ -252,6 +276,17 @@ static int add_signal( return -1; } +static void on_reply_j(int status, const char *reply, struct jrespw *jrespw) +{ + struct json_object *obj; + + if (!jparse(jrespw->jbus, reply, &obj)) + obj = json_object_new_string(reply); + jrespw->onresp(status, obj, jrespw->data); + json_object_put(obj); + free(jrespw); +} + /* * Creates a message for 'method' with one string parameter being 'query' * and sends it to the destination, object and interface linked to 'jbus'. @@ -269,359 +304,37 @@ static int call( void (*onresp_j) (int, struct json_object *, void *), void *data) { - DBusMessage *msg; struct jrespw *resp; + if (onresp_j == NULL) + return sbus_call(jbus->sbus, jbus->name, jbus->path, jbus->name, + method, query, onresp_s, data); + /* allocates the response structure */ resp = malloc(sizeof *resp); - if (resp == NULL) { - errno = ENOMEM; + if (resp == NULL) goto error; - } - - /* creates the message */ - msg = dbus_message_new_method_call(jbus->name, jbus->path, jbus->name, - method); - if (msg == NULL) { - errno = ENOMEM; - goto error2; - } - - /* fill it */ - if (!dbus_message_append_args - (msg, DBUS_TYPE_STRING, &query, DBUS_TYPE_INVALID)) { - errno = ENOMEM; - goto error3; - } - - /* send it */ - if (!dbus_connection_send(jbus->connection, msg, &resp->serial)) { - /* TODO: which error? */ - goto error3; - } - - /* release the message that is not more used */ - dbus_message_unref(msg); /* fulfill the response structure */ + resp->jbus = jbus; + resp->onresp = onresp_j; resp->data = data; - resp->onresp_s = onresp_s; - resp->onresp_j = onresp_j; + if (sbus_call(jbus->sbus, jbus->name, jbus->path, jbus->name, + method, query, (void*)on_reply_j, resp)) + goto error2; - /* links the response to list of reponse waiters */ - resp->next = jbus->waiters; - jbus->waiters = resp; return 0; - error3: - dbus_message_unref(msg); error2: free(resp); error: + errno = ENOMEM; return -1; } -/* - * Callback function for synchronous calls. - * This function fills the respsync structure pointed by 'data' - * with the copy of the answer. - */ -static void sync_of_replies(int status, const char *value, void *data) -{ - struct respsync *s = data; - s->value = status ? NULL : strdup(value ? value : ""); - s->replied = 1; -} - -/* - * Parses the json-string 'msg' to create a json object stored - * in 'obj'. It uses the tokener of 'jbus'. This is a small - * improvement to avoid recreation of tokeners. - * - * Returns 1 in case of success and put the result in *'obj'. - * Returns 0 in case of error and put NULL in *'obj'. - */ -static int jparse(struct jbus *jbus, const char *msg, struct json_object **obj) -{ - json_tokener_reset(jbus->tokener); - *obj = json_tokener_parse_ex(jbus->tokener, msg, -1); - if (json_tokener_get_error(jbus->tokener) == json_tokener_success) - return 1; - json_object_put(*obj); - *obj = NULL; - return 0; -} - -/*********************** STATIC DBUS MESSAGE HANDLING *****************/ - -/* - * Handles incomming responses 'message' on 'jbus'. Response are - * either expected if 'iserror' == 0 or errors if 'iserror' != 0. - * - * Returns DBUS_HANDLER_RESULT_HANDLED or DBUS_HANDLER_RESULT_NOT_YET_HANDLED - * as defined by the dbus function 'dbus_connection_add_filter'. - */ -static DBusHandlerResult incoming_resp( - struct jbus *jbus, - DBusMessage * message, - int iserror) -{ - int status; - const char *str; - struct jrespw *jrw, **prv; - struct json_object *reply; - dbus_uint32_t serial; - - /* search for the waiter */ - serial = dbus_message_get_reply_serial(message); - prv = &jbus->waiters; - while ((jrw = *prv) != NULL && jrw->serial != serial) - prv = &jrw->next; - if (jrw == NULL) - return DBUS_HANDLER_RESULT_NOT_YET_HANDLED; - *prv = jrw->next; - - /* retrieve the string value */ - if (dbus_message_get_args - (message, NULL, DBUS_TYPE_STRING, &str, DBUS_TYPE_INVALID)) - status = 0; - else { - status = -1; - str = NULL; - reply = NULL; - } - - /* treat it */ - if (jrw->onresp_s) - jrw->onresp_s(iserror ? -1 : status, str, jrw->data); - else { - status = jparse(jbus, str, &reply) - 1; - jrw->onresp_j(iserror ? -1 : status, reply, jrw->data); - json_object_put(reply); - } - - free(jrw); - return DBUS_HANDLER_RESULT_HANDLED; -} - -/* - * Handles incomming on 'jbus' method calls for 'message'. - * - * Returns DBUS_HANDLER_RESULT_HANDLED or DBUS_HANDLER_RESULT_NOT_YET_HANDLED - * as defined by the dbus function 'dbus_connection_add_filter'. - */ -static DBusHandlerResult incoming_call( - struct jbus *jbus, - DBusMessage * message) -{ - struct jservice *srv; - struct jreq *jreq; - const char *str; - const char *method; - struct json_object *query; - - /* search for the service */ - if (!matchitf(jbus, message)) - return DBUS_HANDLER_RESULT_NOT_YET_HANDLED; - method = dbus_message_get_member(message); - if (method == NULL) - return DBUS_HANDLER_RESULT_NOT_YET_HANDLED; - srv = jbus->services; - while (srv != NULL && strcmp(method, srv->method)) - srv = srv->next; - if (srv == NULL) - return DBUS_HANDLER_RESULT_NOT_YET_HANDLED; - - /* creates and init the jreq structure */ - jreq = malloc(sizeof *jreq); - if (jreq == NULL) - return DBUS_HANDLER_RESULT_NEED_MEMORY; - jreq->request = dbus_message_ref(message); - jreq->connection = dbus_connection_ref(jbus->connection); - - /* retrieve the string parameter of the message */ - if (!dbus_message_get_args - (message, NULL, DBUS_TYPE_STRING, &str, DBUS_TYPE_INVALID)) - goto invalid_request; - - /* send the message to the callback */ - if (srv->oncall_s) { - /* handling strings only */ - srv->oncall_s(jreq, str, srv->data); - } else { - /* handling json only */ - if (!jparse(jbus, str, &query)) - goto invalid_request; - srv->oncall_j(jreq, query, srv->data); - json_object_put(query); - } - return DBUS_HANDLER_RESULT_HANDLED; - -invalid_request: - jbus_reply_error_s(jreq, invalid_request_string); - return DBUS_HANDLER_RESULT_HANDLED; -} - -/* - * Handles incomming on 'jbus' signal propagated with 'message'. - * - * This is a design choice to ignore invalid signals. - * - * Returns DBUS_HANDLER_RESULT_HANDLED or DBUS_HANDLER_RESULT_NOT_YET_HANDLED - * as defined by the dbus function 'dbus_connection_add_filter'. - */ -static DBusHandlerResult incoming_signal( - struct jbus *jbus, - DBusMessage * message) -{ - struct jsignal *sig; - const char *str; - const char *name; - struct json_object *obj; - - /* search for the signal name */ - if (!matchitf(jbus, message)) - return DBUS_HANDLER_RESULT_NOT_YET_HANDLED; - name = dbus_message_get_member(message); - if (name == NULL) - return DBUS_HANDLER_RESULT_NOT_YET_HANDLED; - sig = jbus->signals; - while (sig != NULL && strcmp(name, sig->name)) - sig = sig->next; - if (sig == NULL) - return DBUS_HANDLER_RESULT_NOT_YET_HANDLED; - - /* retrieve the string value */ - if (dbus_message_get_args - (message, NULL, DBUS_TYPE_STRING, &str, DBUS_TYPE_INVALID)) { - if (sig->onsignal_s) { - /* handling strings only */ - sig->onsignal_s(str, sig->data); - } else { - /* handling json only (if valid) */ - if (jparse(jbus, str, &obj)) { - sig->onsignal_j(obj, sig->data); - json_object_put(obj); - } - } - } - return DBUS_HANDLER_RESULT_HANDLED; -} - -/* - * Filters incomming messages as defined by the dbus function - * 'dbus_connection_add_filter'. - * Returns DBUS_HANDLER_RESULT_HANDLED or DBUS_HANDLER_RESULT_NOT_YET_HANDLED. - */ -static DBusHandlerResult incoming( - DBusConnection * connection, - DBusMessage * message, - void *data) -{ - struct jbus *jbus = data; - switch (dbus_message_get_type(message)) { - case DBUS_MESSAGE_TYPE_METHOD_CALL: - return incoming_call(jbus, message); - case DBUS_MESSAGE_TYPE_METHOD_RETURN: - return incoming_resp(jbus, message, 0); - case DBUS_MESSAGE_TYPE_ERROR: - return incoming_resp(jbus, message, 1); - case DBUS_MESSAGE_TYPE_SIGNAL: - return incoming_signal(jbus, message); - } - return DBUS_HANDLER_RESULT_NOT_YET_HANDLED; -} - -/*********************** STATIC DBUS WATCH/POLLING INTERFACE **********/ - -/* - * Set the watched flags of 'jbus' following what DBUS expects by 'watch' - */ -static void watchset(DBusWatch * watch, struct jbus *jbus) -{ - unsigned int flags; - short wf; - - flags = dbus_watch_get_flags(watch); - wf = jbus->watchflags; - if (dbus_watch_get_enabled(watch)) { - 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; -} - -/* - * DBUS Callback for removing a 'watch'. - * See function 'dbus_connection_set_watch_functions' - */ -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--; -} - -/* - * DBUS Callback for changing a 'watch'. - * See function 'dbus_connection_set_watch_functions' - */ -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); -} - -/* - * DBUS Callback for adding a 'watch'. - * See function 'dbus_connection_set_watch_functions' - */ -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 *****************************************/ /* - * Creates a 'jbus' bound to DBUS system using 'path' and returns it. - * See 'create_jbus' - */ -struct jbus *create_jbus_system(const char *path) -{ - return create_jbus(path, 0); -} - -/* - * Creates a 'jbus' bound to DBUS session using 'path' and returns it. - * See 'create_jbus' - */ -struct jbus *create_jbus_session(const char *path) -{ - return create_jbus(path, 1); -} - -/* * Creates a 'jbus' bound the 'path' and it derived names and linked * either to the DBUS SYSTEM when 'session' is nul or to the DBUS SESSION * if 'session' is not nul. @@ -638,7 +351,7 @@ struct jbus *create_jbus_session(const char *path) * * Returns the created jbus or NULL in case of error. */ -struct jbus *create_jbus(const char *path, int session) +struct jbus *create_jbus(struct sbus *sbus, const char *path) { struct jbus *jbus; char *name; @@ -687,14 +400,7 @@ struct jbus *create_jbus(const char *path, int session) } /* connect and init */ - jbus->connection = dbus_bus_get(session ? DBUS_BUS_SESSION - : DBUS_BUS_SYSTEM, 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; + jbus->sbus = sbus; return jbus; @@ -719,16 +425,25 @@ void jbus_addref(struct jbus *jbus) void jbus_unref(struct jbus *jbus) { struct jservice *srv; + struct jsignal *sig; if (!--jbus->refcount) { - if (jbus->connection != NULL) - dbus_connection_unref(jbus->connection); while ((srv = jbus->services) != NULL) { jbus->services = srv->next; free(srv->method); free(srv); } + while ((sig = jbus->signals) != NULL) { + jbus->signals = sig->next; + free(sig->name); + free(sig); + } + if (jbus->sservice != NULL) + sbus_remove_service(jbus->sbus, jbus->sservice); + if (jbus->ssignal != NULL) + sbus_remove_signal(jbus->sbus, jbus->ssignal); if (jbus->tokener != NULL) json_tokener_free(jbus->tokener); + sbus_unref(jbus->sbus); free(jbus->name); free(jbus->path); free(jbus); @@ -736,79 +451,49 @@ void jbus_unref(struct jbus *jbus) } /* - * Replies an error of string 'error' to the request handled by 'jreq'. - * Also destroys the request 'jreq' that must not be used later. + * Replies an error of string 'error' to the request handled by 'smsg'. + * Also destroys the request 'smsg' that must not be used later. * * Returns 0 in case of success or -1 in case of error. */ -int jbus_reply_error_s(struct jreq *jreq, const char *error) +int jbus_reply_error_s(struct sbusmsg *smsg, const char *error) { - int rc = -1; - DBusMessage *message; - - message = dbus_message_new_error(jreq->request, DBUS_ERROR_FAILED, - error); - if (message == NULL) - errno = ENOMEM; - else { - if (dbus_connection_send(jreq->connection, message, NULL)) - rc = 0; - dbus_message_unref(message); - } - free_jreq(jreq); - return rc; + return sbus_reply_error(smsg, error); } /* - * Replies an error of json 'reply' to the request handled by 'jreq'. - * Also destroys the request 'jreq' that must not be used later. + * Replies an error of json 'reply' to the request handled by 'smsg'. + * Also destroys the request 'smsg' that must not be used later. * * Returns 0 in case of success or -1 in case of error. */ -int jbus_reply_error_j(struct jreq *jreq, struct json_object *reply) +int jbus_reply_error_j(struct sbusmsg *smsg, struct json_object *reply) { const char *str = json_object_to_json_string(reply); - return str ? jbus_reply_error_s(jreq, str) : reply_out_of_memory(jreq); + return str ? jbus_reply_error_s(smsg, str) : reply_out_of_memory(smsg); } /* - * Replies normally the string 'reply' to the request handled by 'jreq'. - * Also destroys the request 'jreq' that must not be used later. + * Replies normally the string 'reply' to the request handled by 'smsg'. + * Also destroys the request 'smsg' that must not be used later. * * Returns 0 in case of success or -1 in case of error. */ -int jbus_reply_s(struct jreq *jreq, const char *reply) +int jbus_reply_s(struct sbusmsg *smsg, const char *reply) { - int rc = -1; - DBusMessage *message; - - message = dbus_message_new_method_return(jreq->request); - if (message == NULL) - return reply_out_of_memory(jreq); - - if (!dbus_message_append_args - (message, DBUS_TYPE_STRING, &reply, DBUS_TYPE_INVALID)) { - dbus_message_unref(message); - return reply_out_of_memory(jreq); - } - - if (dbus_connection_send(jreq->connection, message, NULL)) - rc = 0; - dbus_message_unref(message); - free_jreq(jreq); - return rc; + return sbus_reply(smsg, reply); } /* - * Replies normally the json 'reply' to the request handled by 'jreq'. - * Also destroys the request 'jreq' that must not be used later. + * Replies normally the json 'reply' to the request handled by 'smsg'. + * Also destroys the request 'smsg' that must not be used later. * * Returns 0 in case of success or -1 in case of error. */ -int jbus_reply_j(struct jreq *jreq, struct json_object *reply) +int jbus_reply_j(struct sbusmsg *smsg, struct json_object *reply) { const char *str = json_object_to_json_string(reply); - return str ? jbus_reply_s(jreq, str) : reply_out_of_memory(jreq); + return str ? jbus_reply_s(smsg, str) : reply_out_of_memory(smsg); } /* @@ -818,28 +503,7 @@ int jbus_reply_j(struct jreq *jreq, struct json_object *reply) */ int jbus_send_signal_s(struct jbus *jbus, const char *name, const char *content) { - int rc = -1; - DBusMessage *message; - - message = dbus_message_new_signal(jbus->path, jbus->name, name); - if (message == NULL) - goto error; - - 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; - } - - if (dbus_connection_send(jbus->connection, message, NULL)) - rc = 0; - dbus_message_unref(message); - return rc; - - error: - errno = ENOMEM; - return -1; + return sbus_send_signal(jbus->sbus, jbus->name, jbus->path, jbus->name, name, content); } /* @@ -864,7 +528,7 @@ int jbus_send_signal_j(struct jbus *jbus, const char *name, * * The callback 'oncall' is invoked for handling incoming method * calls. It receives 3 parameters: - * 1. struct jreq *: a handler to data to be used for replying + * 1. struct sbusmsg *: a handler to data to be used for replying * 2. const char *: the received string * 3. void *: the closure 'data' set by this function * @@ -873,7 +537,7 @@ int jbus_send_signal_j(struct jbus *jbus, const char *name, int jbus_add_service_s( struct jbus *jbus, const char *method, - void (*oncall) (struct jreq *, const char *, void *), + void (*oncall) (struct sbusmsg *, const char *, void *), void *data) { return add_service(jbus, method, oncall, NULL, data); @@ -885,7 +549,7 @@ int jbus_add_service_s( * * The callback 'oncall' is invoked for handling incoming method * calls. It receives 3 parameters: - * 1. struct jreq *: a handler to data to be used for replying + * 1. struct sbusmsg *: a handler to data to be used for replying * 2. struct json_object *: the received json * 3. void *: the closure 'data' set by this function * @@ -894,7 +558,7 @@ int jbus_add_service_s( int jbus_add_service_j( struct jbus *jbus, const char *method, - void (*oncall) (struct jreq *, struct json_object *, void *), + void (*oncall) (struct sbusmsg *, struct json_object *, void *), void *data) { return add_service(jbus, method, NULL, oncall, data); @@ -911,146 +575,7 @@ int jbus_add_service_j( */ int jbus_start_serving(struct jbus *jbus) { - int status = dbus_bus_request_name(jbus->connection, jbus->name, - DBUS_NAME_FLAG_DO_NOT_QUEUE, NULL); - switch (status) { - case DBUS_REQUEST_NAME_REPLY_PRIMARY_OWNER: - case DBUS_REQUEST_NAME_REPLY_ALREADY_OWNER: - return 0; - case DBUS_REQUEST_NAME_REPLY_EXISTS: - case DBUS_REQUEST_NAME_REPLY_IN_QUEUE: - default: - errno = EADDRINUSE; - return -1; - } -} - -/* - * Fills the at least 'njbuses' structures of array 'fds' with data needed - * to poll the 'njbuses' buses pointed by 'jbuses'. - * - * Returns the count of 'fds' structures filled. - */ -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; -} - -/* - * Dispatchs a maximum of 'maxcount' events received by poll in 'fds' for the - * 'njbuses' jbuses of the array 'jbuses'. - * - * Returns the count of event dispatched. - */ -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; -} - -/* - * Dispatches 'maxcount' of buffered data from the 'njbuses' jbuses of the - * array 'jbuses'. - * - * Returns the count of event dispatched. - */ -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); - sts = dbus_connection_get_dispatch_status( - jbuses[i]->connection); - while (sts == DBUS_DISPATCH_DATA_REMAINS && r < maxcount) { - sts = dbus_connection_dispatch(jbuses[i]->connection); - r++; - } - } - return r; -} - -/* - * Polls during at most 'toms' milliseconds and dispatches 'maxcount' - * of events from the 'njbuses' jbuses of the array 'jbuses'. - * - * Returns the count of event dispatched or -1 in case of error. - */ -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((unsigned)njbuses * sizeof *fds); - assert(fds != NULL); - - r = jbus_dispatch_multiple(jbuses, njbuses, maxcount); - if (r) - return r; - n = jbus_fill_pollfds(jbuses, njbuses, fds); - for (;;) { - s = poll(fds, (nfds_t) n, toms); - if (s >= 0) - break; - if (errno != EINTR) - return r ? r : s; - toms = 0; - } - n = jbus_dispatch_pollfds(jbuses, njbuses, fds, maxcount - r); - return n >= 0 ? r + n : r ? r : n; -} - -/* - * Polls during at most 'toms' milliseconds and dispatches - * the events from 'jbus'. - * - * Returns the count of event dispatched or -1 in case of error. - */ -int jbus_read_write_dispatch(struct jbus *jbus, int toms) -{ - int r = jbus_read_write_dispatch_multiple(&jbus, 1, toms, 1000); - return r < 0 ? r : 0; + return sbus_add_name(jbus->sbus, jbus->name); } /* @@ -1158,12 +683,8 @@ char *jbus_call_ss_sync( const char *method, const char *query) { - struct respsync synchro; - synchro.value = NULL; - synchro.replied = - jbus_call_ss(jbus, method, query, sync_of_replies, &synchro); - while (!synchro.replied && !jbus_read_write_dispatch(jbus, -1)) ; - return synchro.value; + return sbus_call_sync(jbus->sbus, jbus->name, jbus->path, jbus->name, + method, query); } /* @@ -1264,25 +785,50 @@ int jbus_on_signal_j( /****************** FEW LITTLE TESTS *****************************************/ -#ifdef SERVER +#if defined(SERVER)||defined(CLIENT) #include <stdio.h> #include <unistd.h> -struct jbus *jbus; -void ping(struct jreq *jreq, struct json_object *request, void *unused) +#include "utils-upoll.h" + +static int mwait(int timeout, void *closure) +{ + upoll_wait(-1); + return 0; +} + +static const struct sbus_itf uitf = { + .wait = (void*)mwait, + .open = (void*)upoll_open, + .on_readable = (void*)upoll_on_readable, + .on_writable = (void*)upoll_on_writable, + .on_hangup = (void*)upoll_on_hangup, + .close = (void*)upoll_close +}; + +static struct sbus *sbus; +static struct jbus *jbus; + +static struct sbus *msbus() +{ + return sbus ? : (sbus = sbus_session(&uitf, NULL)); +} + +#ifdef SERVER +void ping(struct sbusmsg *smsg, struct json_object *request, void *unused) { printf("ping(%s) -> %s\n", json_object_to_json_string(request), json_object_to_json_string(request)); - jbus_reply_j(jreq, request); + jbus_reply_j(smsg, request); json_object_put(request); } -void incr(struct jreq *jreq, struct json_object *request, void *unused) +void incr(struct sbusmsg *smsg, struct json_object *request, void *unused) { static int counter = 0; struct json_object *res = json_object_new_int(++counter); printf("incr(%s) -> %s\n", json_object_to_json_string(request), json_object_to_json_string(res)); - jbus_reply_j(jreq, res); + jbus_reply_j(smsg, res); jbus_send_signal_j(jbus, "incremented", res); json_object_put(res); json_object_put(request); @@ -1291,18 +837,17 @@ void incr(struct jreq *jreq, struct json_object *request, void *unused) int main() { int s1, s2, s3; - jbus = create_jbus(1, "/bzh/iot/jdbus"); + jbus = create_jbus(msbus(), "/bzh/iot/jdbus"); s1 = jbus_add_service_j(jbus, "ping", ping, NULL); s2 = jbus_add_service_j(jbus, "incr", incr, NULL); s3 = jbus_start_serving(jbus); printf("started %d %d %d\n", s1, s2, s3); - while (!jbus_read_write_dispatch(jbus, -1)) ; + while (!mwait(-1,jbus)) ; + return 0; } #endif + #ifdef CLIENT -#include <stdio.h> -#include <unistd.h> -struct jbus *jbus; void onresp(int status, struct json_object *response, void *data) { printf("resp: %d, %s, %s\n", status, (char *)data, @@ -1310,25 +855,28 @@ void onresp(int status, struct json_object *response, void *data) json_object_put(response); } -void signaled(const char *data) +void signaled(const char *content, void *data) { - printf("signaled with {%s}\n", data); + printf("signaled with {%s}/%s\n", content, (char*)data); } int main() { int i = 10; - jbus = create_jbus(1, "/bzh/iot/jdbus"); - jbus_on_signal_s(jbus, "incremented", signaled); + jbus = create_jbus(msbus(), "/bzh/iot/jdbus"); + jbus_on_signal_s(jbus, "incremented", signaled, "closure-signal"); while (i--) { jbus_call_sj(jbus, "ping", "{\"toto\":[1,2,3,4,true,\"toto\"]}", onresp, "ping"); jbus_call_sj(jbus, "incr", "{\"doit\":\"for-me\"}", onresp, "incr"); - jbus_read_write_dispatch(jbus, 1); + mwait(-1,jbus); } printf("[[[%s]]]\n", jbus_call_ss_sync(jbus, "ping", "\"formidable!\"")); - while (!jbus_read_write_dispatch(jbus, -1)) ; + while (!mwait(-1,jbus)) ; + return 0; } #endif +#endif + |