diff options
| -rw-r--r-- | include/afb/afb-event-itf.h | 84 | ||||
| -rw-r--r-- | include/afb/afb-plugin.h | 17 | ||||
| -rw-r--r-- | include/afb/afb-req-itf.h | 28 | ||||
| -rw-r--r-- | src/CMakeLists.txt | 1 | ||||
| -rw-r--r-- | src/afb-api-dbus.c | 23 | ||||
| -rw-r--r-- | src/afb-api-dbus.h | 4 | ||||
| -rw-r--r-- | src/afb-api-so.c | 26 | ||||
| -rw-r--r-- | src/afb-context.c | 1 | ||||
| -rw-r--r-- | src/afb-evt.c | 261 | ||||
| -rw-r--r-- | src/afb-evt.h | 36 | ||||
| -rw-r--r-- | src/afb-hreq.c | 15 | ||||
| -rw-r--r-- | src/afb-hreq.h | 3 | ||||
| -rw-r--r-- | src/afb-ws-json1.c | 38 | ||||
| -rw-r--r-- | src/afb-ws-json1.h | 3 | ||||
| -rw-r--r-- | src/session.c | 108 | ||||
| -rw-r--r-- | src/session.h | 16 |
16 files changed, 500 insertions, 164 deletions
diff --git a/include/afb/afb-event-itf.h b/include/afb/afb-event-itf.h new file mode 100644 index 00000000..47ffa387 --- /dev/null +++ b/include/afb/afb-event-itf.h @@ -0,0 +1,84 @@ +/* + * Copyright (C) 2016 "IoT.bzh" + * Author: José Bollo <jose.bollo@iot.bzh> + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#pragma once + +/* avoid inclusion of <json-c/json.h> */ +struct json_object; + +/* + * Interface for handling requests. + * It records the functions to be called for the request. + * Don't use this structure directly. + * Use the helper functions documented below. + */ +struct afb_event_itf { + /* CAUTION: respect the order, add at the end */ + + int (*broadcast)(void *closure, struct json_object *obj); + int (*push)(void *closure, struct json_object *obj); + void (*drop)(void *closure); +}; + +/* + * Describes the request by plugins from afb-daemon + */ +struct afb_event { + const struct afb_event_itf *itf; /* the interface to use */ + void *closure; /* the closure argument for functions of 'itf' */ +}; + +/* + * Broadcasts widely the 'event' with the data 'object'. + * 'object' can be NULL. + * + * For conveniency, the function calls 'json_object_put' for 'object'. + * Thus, in the case where 'object' should remain available after + * the function returns, the function 'json_object_get' shall be used. + * + * Returns the count of clients that received the event. + */ +static inline int afb_event_broadcast(struct afb_event event, struct json_object *object) +{ + return event.itf->broadcast(event.closure, object); +} + +/* + * Pushes the 'event' with the data 'object' to its obeservers. + * 'object' can be NULL. + * + * For conveniency, the function calls 'json_object_put' for 'object'. + * Thus, in the case where 'object' should remain available after + * the function returns, the function 'json_object_get' shall be used. + * + * Returns the count of clients that received the event. + */ +static inline int afb_event_push(struct afb_event event, struct json_object *object) +{ + return event.itf->push(event.closure, object); +} + +/* + * Drops the data associated to the event + * After calling this function, the event + * MUST NOT BE USED ANYMORE. + */ +static inline void afb_event_drop(struct afb_event event) +{ + event.itf->drop(event.closure); +} + diff --git a/include/afb/afb-plugin.h b/include/afb/afb-plugin.h index 1eb3475b..85e8de7e 100644 --- a/include/afb/afb-plugin.h +++ b/include/afb/afb-plugin.h @@ -38,6 +38,7 @@ * Some function of the library are exported to afb-daemon. */ +#include <afb/afb-event-itf.h> #include <afb/afb-req-itf.h> /* @@ -140,11 +141,12 @@ struct sd_bus; * Definition of the facilities provided by the daemon. */ struct afb_daemon_itf { - void (*event_broadcast)(void *closure, const char *name, struct json_object *object); /* broadcasts evant 'name' with 'object' */ + int (*event_broadcast)(void *closure, const char *name, struct json_object *object); /* broadcasts evant 'name' with 'object' */ struct sd_event *(*get_event_loop)(void *closure); /* gets the common systemd's event loop */ struct sd_bus *(*get_user_bus)(void *closure); /* gets the common systemd's user d-bus */ struct sd_bus *(*get_system_bus)(void *closure); /* gets the common systemd's system d-bus */ void (*vverbose)(void*closure, int level, const char *file, int line, const char *fmt, va_list args); + struct afb_event (*event_make)(void *closure, const char *name); /* creates an event of 'name' */ }; /* @@ -206,13 +208,24 @@ static inline struct sd_bus *afb_daemon_get_system_bus(struct afb_daemon daemon) * For conveniency, the function calls 'json_object_put' for 'object'. * Thus, in the case where 'object' should remain available after * the function returns, the function 'json_object_get' shall be used. + * + * Returns the count of clients that received the event. */ -static inline void afb_daemon_broadcast_event(struct afb_daemon daemon, const char *name, struct json_object *object) +static inline int afb_daemon_broadcast_event(struct afb_daemon daemon, const char *name, struct json_object *object) { return daemon.itf->event_broadcast(daemon.closure, name, object); } /* + * Creates an event of 'name' and returns it. + * 'daemon' MUST be the daemon given in interface when activating the plugin. + */ +static inline struct afb_event afb_daemon_make_event(struct afb_daemon daemon, const char *name) +{ + return daemon.itf->event_make(daemon.closure, name); +} + +/* * Send a message described by 'fmt' and following parameters * to the journal for the verbosity 'level'. * 'file' and 'line' are indicators of position of the code in source files. diff --git a/include/afb/afb-req-itf.h b/include/afb/afb-req-itf.h index f4fab551..2b3bc467 100644 --- a/include/afb/afb-req-itf.h +++ b/include/afb/afb-req-itf.h @@ -25,6 +25,8 @@ #include <stdarg.h> #include <stdio.h> +#include <afb/afb-event-itf.h> + /* avoid inclusion of <json-c/json.h> */ struct json_object; @@ -65,6 +67,9 @@ struct afb_req_itf { void (*session_close)(void *closure); int (*session_set_LOA)(void *closure, unsigned level); + + int (*subscribe)(void *closure, struct afb_event event); + int (*unsubscribe)(void *closure, struct afb_event event); }; /* @@ -315,6 +320,29 @@ static inline struct afb_req afb_req_unstore(struct afb_req *req) return result; } +/* + * Establishes for the client link identified by 'req' a subscription + * to the 'event'. + * Returns 0 in case of successful subscription or -1 in case of error. + */ +static inline int afb_req_subscribe(struct afb_req req, struct afb_event event) +{ + return req.itf->subscribe(req.closure, event); +} + +/* + * Revokes the subscription established to the 'event' for the client + * link identified by 'req'. + * Returns 0 in case of successful subscription or -1 in case of error. + */ +static inline int afb_req_unsubscribe(struct afb_req req, struct afb_event event) +{ + return req.itf->unsubscribe(req.closure, event); +} + + + + /* internal use */ static inline const char *afb_req_raw(struct afb_req req, size_t *size) { diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index ca20665b..cbaf286f 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -35,6 +35,7 @@ ADD_LIBRARY(afb-lib STATIC afb-apis.c afb-common.c afb-context.c + afb-evt.c afb-hreq.c afb-hsrv.c afb-hswitch.c diff --git a/src/afb-api-dbus.c b/src/afb-api-dbus.c index c0bc57a8..eda7985f 100644 --- a/src/afb-api-dbus.c +++ b/src/afb-api-dbus.c @@ -35,6 +35,7 @@ #include "afb-apis.h" #include "afb-api-so.h" #include "afb-context.h" +#include "afb-evt.h" #include "verbose.h" static const char DEFAULT_PATH_PREFIX[] = "/org/agl/afb/api/"; @@ -52,6 +53,7 @@ struct api_dbus char *path; /* path of the object for the API */ char *name; /* name/interface of the object */ char *api; /* api name of the interface */ + struct afb_evt_listener *listener; }; #define RETOK 1 @@ -302,7 +304,7 @@ static int api_dbus_client_on_event(sd_bus_message *m, void *userdata, sd_bus_er ERROR("unreadable event"); else { object = json_tokener_parse(data); - ctxClientEventSend(NULL, event, object); + afb_evt_broadcast(event, object); json_object_put(object); } return 1; @@ -444,7 +446,7 @@ static void dbus_req_send(struct dbus_req *dreq, const char *buffer, size_t size dbus_req_reply(dreq, RETRAW, buffer, ""); } -struct afb_req_itf dbus_req_itf = { +const struct afb_req_itf afb_api_dbus_req_itf = { .json = (void*)dbus_req_json, .get = (void*)dbus_req_get, .success = (void*)dbus_req_success, @@ -504,7 +506,7 @@ static int api_dbus_server_on_object_called(sd_bus_message *message, void *userd dreq->message = sd_bus_message_ref(message); dreq->json = NULL; dreq->refcount = 1; - areq.itf = &dbus_req_itf; + areq.itf = &afb_api_dbus_req_itf; areq.closure = dreq; afb_apis_call_(areq, &dreq->context, api->api, method); dbus_req_unref(dreq); @@ -522,19 +524,6 @@ static void afb_api_dbus_server_send_event(struct api_dbus *api, const char *eve json_object_put(object); } -static int afb_api_dbus_server_expects_event(struct api_dbus *api, const char *event) -{ - size_t len = strlen(api->api); - if (strncasecmp(event, api->api, len) != 0) - return 0; - return event[len] == '.'; -} - -static struct afb_event_listener_itf evitf = { - .send = (void*)afb_api_dbus_server_send_event, - .expects = (void*)afb_api_dbus_server_expects_event -}; - /* create the service */ int afb_api_dbus_add_server(const char *path) { @@ -563,7 +552,7 @@ int afb_api_dbus_add_server(const char *path) } INFO("afb service over dbus installed, name %s, path %s", api->name, api->path); - ctxClientEventListenerAdd(NULL, (struct afb_event_listener){ .itf = &evitf, .closure = api }); + api->listener = afb_evt_listener_create((void*)afb_api_dbus_server_send_event, api); return 0; error3: diff --git a/src/afb-api-dbus.h b/src/afb-api-dbus.h index c8a7bc3f..10f5f7ff 100644 --- a/src/afb-api-dbus.h +++ b/src/afb-api-dbus.h @@ -18,6 +18,10 @@ #pragma once +struct afb_req_itf; + +extern const struct afb_req_itf afb_api_dbus_req_itf; + extern int afb_api_dbus_add_client(const char *path); extern int afb_api_dbus_add_server(const char *path); diff --git a/src/afb-api-so.c b/src/afb-api-so.c index b741b13a..84f69753 100644 --- a/src/afb-api-so.c +++ b/src/afb-api-so.c @@ -30,6 +30,7 @@ #include <afb/afb-plugin.h> #include <afb/afb-req-itf.h> +#include <afb/afb-event-itf.h> #include "session.h" #include "afb-common.h" @@ -37,6 +38,7 @@ #include "afb-apis.h" #include "afb-api-so.h" #include "afb-sig-handler.h" +#include "afb-evt.h" #include "verbose.h" /* @@ -58,18 +60,37 @@ void afb_api_so_set_timeout(int to) api_timeout = to; } +static struct afb_event afb_api_so_event_make(struct api_so_desc *desc, const char *name) +{ + size_t length; + char *event; + + /* makes the event name */ + assert(desc->plugin != NULL); + length = strlen(name); + event = alloca(length + 2 + desc->apilength); + memcpy(event, desc->plugin->v1.prefix, desc->apilength); + event[desc->apilength] = '/'; + memcpy(event + desc->apilength + 1, name, length + 1); + + /* crate the event */ + return afb_evt_create_event(event); +} + static int afb_api_so_event_broadcast(struct api_so_desc *desc, const char *name, struct json_object *object) { size_t length; char *event; + /* makes the event name */ assert(desc->plugin != NULL); length = strlen(name); event = alloca(length + 2 + desc->apilength); memcpy(event, desc->plugin->v1.prefix, desc->apilength); event[desc->apilength] = '/'; memcpy(event + desc->apilength + 1, name, length + 1); - return ctxClientEventSend(NULL, event, object); + + return afb_evt_broadcast(event, object); } static void afb_api_so_vverbose(struct api_so_desc *desc, int level, const char *file, int line, const char *fmt, va_list args) @@ -89,7 +110,8 @@ static const struct afb_daemon_itf daemon_itf = { .get_event_loop = (void*)afb_common_get_event_loop, .get_user_bus = (void*)afb_common_get_user_bus, .get_system_bus = (void*)afb_common_get_system_bus, - .vverbose = (void*)afb_api_so_vverbose + .vverbose = (void*)afb_api_so_vverbose, + .event_make = (void*)afb_api_so_event_make }; struct monitoring { diff --git a/src/afb-context.c b/src/afb-context.c index ba093c37..5fe32764 100644 --- a/src/afb-context.c +++ b/src/afb-context.c @@ -1,6 +1,7 @@ /* * Copyright (C) 2015, 2016 "IoT.bzh" * Author "Fulup Ar Foll" + * Author José Bollo <jose.bollo@iot.bzh> * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. diff --git a/src/afb-evt.c b/src/afb-evt.c new file mode 100644 index 00000000..00261b6c --- /dev/null +++ b/src/afb-evt.c @@ -0,0 +1,261 @@ +/* + * Copyright (C) 2015, 2016 "IoT.bzh" + * Author "Fulup Ar Foll" + * Author José Bollo <jose.bollo@iot.bzh> + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#define _GNU_SOURCE + +#include <stdlib.h> +#include <string.h> +#include <assert.h> +#include <errno.h> + +#include <json-c/json.h> +#include <afb/afb-event-itf.h> + +#include "afb-evt.h" + + +struct afb_evt_watch; + +struct afb_evt_listener { + struct afb_evt_listener *next; + void (*send)(void *closure, const char *event, struct json_object *object); + void *closure; + struct afb_evt_watch *watchs; + int refcount; +}; + +struct afb_evt_event { + struct afb_evt_watch *watchs; + char name[1]; +}; + +struct afb_evt_watch { + struct afb_evt_event *event; + struct afb_evt_watch *next_by_event; + struct afb_evt_listener *listener; + struct afb_evt_watch *next_by_listener; +}; + +static int evt_broadcast(struct afb_evt_event *evt, struct json_object *obj); +static int evt_push(struct afb_evt_event *evt, struct json_object *obj); +static void evt_drop(struct afb_evt_event *evt); + +static struct afb_event_itf afb_evt_event_itf = { + .broadcast = (void*)evt_broadcast, + .push = (void*)evt_push, + .drop = (void*)evt_drop +}; + +static struct afb_evt_listener *listeners = NULL; + +static inline int evt_trash(struct json_object *obj) +{ + return 0; +} + +static int evt_broadcast(struct afb_evt_event *evt, struct json_object *object) +{ + return afb_evt_broadcast(evt->name, object); +} + +int afb_evt_broadcast(const char *event, struct json_object *object) +{ + int result; + struct afb_evt_listener *listener; + + result = 0; + listener = listeners; + while(listener) { + listener->send(listener->closure, event, json_object_get(object)); + listener = listener->next; + } + json_object_put(object); + return result; +} + +static int evt_push(struct afb_evt_event *evt, struct json_object *obj) +{ + int result; + struct afb_evt_watch *watch; + struct afb_evt_listener *listener; + + result = 0; + watch = evt->watchs; + while(listener) { + listener = watch->listener; + listener->send(listener->closure, evt->name, json_object_get(obj)); + watch = watch->next_by_event; + } + json_object_put(obj); + return result; +} + +static void remove_watch(struct afb_evt_watch *watch) +{ + struct afb_evt_watch **prv; + + prv = &watch->event->watchs; + while(*prv != watch) + prv = &(*prv)->next_by_event; + *prv = watch->next_by_event; + + prv = &watch->listener->watchs; + while(*prv != watch) + prv = &(*prv)->next_by_listener; + *prv = watch->next_by_listener; + + free(watch); +} + +static void evt_drop(struct afb_evt_event *evt) +{ + if (evt != NULL) { + while(evt->watchs != NULL) + remove_watch(evt->watchs); + free(evt); + } +} + +struct afb_event afb_evt_create_event(const char *name) +{ + size_t len; + struct afb_evt_event *evt; + + len = strlen(name); + evt = malloc(len + sizeof * evt); + if (evt != NULL) { + evt->watchs = NULL; + memcpy(evt->name, name, len + 1); + } + return (struct afb_event){ .itf = &afb_evt_event_itf, .closure = evt }; +} + +struct afb_evt_listener *afb_evt_listener_create(void (*send)(void *closure, const char *event, struct json_object *object), void *closure) +{ + struct afb_evt_listener *listener; + + /* search if an instance already exists */ + listener = listeners; + while (listener != NULL) { + if (listener->send == send && listener->closure == closure) + return afb_evt_listener_addref(listener); + listener = listener->next; + } + + /* allocates */ + listener = calloc(1, sizeof *listener); + if (listener != NULL) { + /* init */ + listener->next = listeners; + listener->send = send; + listener->closure = closure; + listener->watchs = NULL; + listener->refcount = 1; + listeners = listener; + } + return listener; +} + +struct afb_evt_listener *afb_evt_listener_addref(struct afb_evt_listener *listener) +{ + listener->refcount++; + return listener; +} + +void afb_evt_listener_unref(struct afb_evt_listener *listener) +{ + if (0 == --listener->refcount) { + struct afb_evt_listener **prv; + + /* remove the watchers */ + while (listener->watchs != NULL) + remove_watch(listener->watchs); + + /* unlink the listener */ + prv = &listeners; + while (*prv != listener) + prv = &(*prv)->next; + *prv = listener->next; + + /* free the listener */ + free(listener); + } +} + +int afb_evt_add_watch(struct afb_evt_listener *listener, struct afb_event event) +{ + struct afb_evt_watch *watch; + struct afb_evt_event *evt; + + /* check parameter */ + if (event.itf != &afb_evt_event_itf) { + errno = EINVAL; + return -1; + } + + /* search the existing watch */ + watch = listener->watchs; + while(watch != NULL) { + if (watch->event == event.closure) + return 0; + watch = watch->next_by_listener; + } + + /* not found, allocate a new */ + watch = malloc(sizeof *watch); + if (watch == NULL) { + errno = ENOMEM; + return -1; + } + + /* initialise and link */ + evt = event.closure; + watch->event = evt; + watch->next_by_event = evt->watchs; + watch->listener = listener; + watch->next_by_listener = listener->watchs; + evt->watchs = watch; + listener->watchs = watch; + + return 0; +} + +int afb_evt_remove_watch(struct afb_evt_listener *listener, struct afb_event event) +{ + struct afb_evt_watch *watch; + + /* check parameter */ + if (event.itf != &afb_evt_event_itf) { + errno = EINVAL; + return -1; + } + + /* search the existing watch */ + watch = listener->watchs; + while(watch != NULL) { + if (watch->event == event.closure) { + /* found: remove it */ + remove_watch(watch); + break; + } + watch = watch->next_by_listener; + } + return 0; +} + + diff --git a/src/afb-evt.h b/src/afb-evt.h new file mode 100644 index 00000000..8e102546 --- /dev/null +++ b/src/afb-evt.h @@ -0,0 +1,36 @@ +/* + * Copyright (C) 2016 "IoT.bzh" + * Author: José Bollo <jose.bollo@iot.bzh> + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#pragma once + +struct afb_event; +struct AFB_clientCtx; + +struct afb_evt_listener; + +extern struct afb_evt_listener *afb_evt_listener_create(void (*send)(void *closure, const char *event, struct json_object *object), void *closure); + +extern int afb_evt_broadcast(const char *event, struct json_object *object); + +extern struct afb_evt_listener *afb_evt_listener_addref(struct afb_evt_listener *listener); +extern void afb_evt_listener_unref(struct afb_evt_listener *listener); + +extern struct afb_event afb_evt_create_event(const char *name); + +extern int afb_evt_add_watch(struct afb_evt_listener *listener, struct afb_event event); +extern int afb_evt_remove_watch(struct afb_evt_listener *listener, struct afb_event event); + diff --git a/src/afb-hreq.c b/src/afb-hreq.c index 4ca8441e..1e38b412 100644 --- a/src/afb-hreq.c +++ b/src/afb-hreq.c @@ -74,8 +74,9 @@ static void req_fail(struct afb_hreq *hreq, const char *status, const char *info static void req_success(struct afb_hreq *hreq, json_object *obj, const char *info); static const char *req_raw(struct afb_hreq *hreq, size_t *size); static void req_send(struct afb_hreq *hreq, const char *buffer, size_t size); +static int req_subscribe_unsubscribe_error(struct afb_hreq *hreq, struct afb_event event); -static const struct afb_req_itf afb_hreq_itf = { +const struct afb_req_itf afb_hreq_req_itf = { .json = (void*)req_json, .get = (void*)req_get, .success = (void*)req_success, @@ -87,7 +88,9 @@ static const struct afb_req_itf afb_hreq_itf = { .addref = (void*)afb_hreq_addref, .unref = (void*)afb_hreq_unref, .session_close = (void*)afb_context_close, - .session_set_LOA = (void*)afb_context_change_loa + .session_set_LOA = (void*)afb_context_change_loa, + .subscribe = (void*)req_subscribe_unsubscribe_error, + .unsubscribe = (void*)req_subscribe_unsubscribe_error }; static struct hreq_data *get_data(struct afb_hreq *hreq, const char *key, int create) @@ -697,7 +700,7 @@ int afb_hreq_post_add_file(struct afb_hreq *hreq, const char *key, const char *f struct afb_req afb_hreq_to_req(struct afb_hreq *hreq) { - return (struct afb_req){ .itf = &afb_hreq_itf, .closure = hreq }; + return (struct afb_req){ .itf = &afb_hreq_req_itf, .closure = hreq }; } static struct afb_arg req_get(struct afb_hreq *hreq, const char *name) @@ -798,6 +801,12 @@ static void req_success(struct afb_hreq *hreq, json_object *obj, const char *inf req_reply(hreq, MHD_HTTP_OK, "success", info, obj); } +static int req_subscribe_unsubscribe_error(struct afb_hreq *hreq, struct afb_event event) +{ + errno = EINVAL; + return -1; +} + int afb_hreq_init_context(struct afb_hreq *hreq) { const char *uuid; diff --git a/src/afb-hreq.h b/src/afb-hreq.h index 836f4703..772cd677 100644 --- a/src/afb-hreq.h +++ b/src/afb-hreq.h @@ -21,6 +21,9 @@ struct AFB_clientCtx; struct json_object; struct hreq_data; struct afb_hsrv; +struct afb_req_itf; + +extern const struct afb_req_itf afb_hreq_req_itf; struct afb_hreq { /* diff --git a/src/afb-ws-json1.c b/src/afb-ws-json1.c index ffb6b81f..5ef751f1 100644 --- a/src/afb-ws-json1.c +++ b/src/afb-ws-json1.c @@ -32,6 +32,7 @@ #include <afb/afb-req-itf.h> #include "afb-apis.h" #include "afb-context.h" +#include "afb-evt.h" #include "verbose.h" static void aws_on_hangup(struct afb_ws_json1 *ws, struct afb_wsj1 *wsj1); @@ -50,22 +51,13 @@ struct afb_ws_json1 void (*cleanup)(void*); void *cleanup_closure; struct AFB_clientCtx *session; + struct afb_evt_listener *listener; struct afb_wsj1 *wsj1; int new_session; }; static void aws_send_event(struct afb_ws_json1 *ws, const char *event, struct json_object *object); -static const struct afb_event_listener_itf event_listener_itf = { - .send = (void*)aws_send_event, - .expects = NULL -}; - -static inline struct afb_event_listener listener_for(struct afb_ws_json1 *aws) -{ - return (struct afb_event_listener){ .itf = &event_listener_itf, .closure = aws }; -} - struct afb_ws_json1 *afb_ws_json1_create(int fd, struct afb_context *context, void (*cleanup)(void*), void *cleanup_closure) { struct afb_ws_json1 *result; @@ -89,7 +81,8 @@ struct afb_ws_json1 *afb_ws_json1_create(int fd, struct afb_context *context, vo if (result->wsj1 == NULL) goto error3; - if (0 > ctxClientEventListenerAdd(result->session, listener_for(result))) + result->listener = afb_evt_listener_create((void*)aws_send_event, result); + if (result->listener == NULL) goto error4; return result; @@ -114,7 +107,7 @@ static struct afb_ws_json1 *aws_addref(struct afb_ws_json1 *ws) static void aws_unref(struct afb_ws_json1 *ws) { if (--ws->refcount == 0) { - ctxClientEventListenerRemove(ws->session, listener_for(ws)); + afb_evt_listener_unref(ws->listener); afb_wsj1_unref(ws->wsj1); if (ws->cleanup != NULL) ws->cleanup(ws->cleanup_closure); @@ -149,9 +142,10 @@ static void wsreq_fail(struct afb_wsreq *wsreq, const char *status, const char * static void wsreq_success(struct afb_wsreq *wsreq, struct json_object *obj, const char *info); static const char *wsreq_raw(struct afb_wsreq *wsreq, size_t *size); static void wsreq_send(struct afb_wsreq *wsreq, const char *buffer, size_t size); +static int wsreq_subscribe(struct afb_wsreq *wsreq, struct afb_event event); +static int wsreq_unsubscribe(struct afb_wsreq *wsreq, struct afb_event event); - -static const struct afb_req_itf wsreq_itf = { +const struct afb_req_itf afb_ws_json1_req_itf = { .json = (void*)wsreq_json, .get = (void*)wsreq_get, .success = (void*)wsreq_success, @@ -163,7 +157,9 @@ static const struct afb_req_itf wsreq_itf = { .addref = (void*)wsreq_addref, .unref = (void*)wsreq_unref, .session_close = (void*)afb_context_close, - .session_set_LOA = (void*)afb_context_change_loa + .session_set_LOA = (void*)afb_context_change_loa, + .subscribe = (void*)wsreq_subscribe, + .unsubscribe = (void*)wsreq_unsubscribe }; static void aws_on_call(struct afb_ws_json1 *ws, const char *api, const char *verb, struct afb_wsj1_msg *msg) @@ -197,7 +193,7 @@ static void aws_on_call(struct afb_ws_json1 *ws, const char *api, const char *ve /* emits the call */ r.closure = wsreq; - r.itf = &wsreq_itf; + r.itf = &afb_ws_json1_req_itf; afb_apis_call_(r, &wsreq->context, api, verb); wsreq_unref(wsreq); } @@ -276,3 +272,13 @@ static void aws_send_event(struct afb_ws_json1 *aws, const char *event, struct j afb_wsj1_send_event_j(aws->wsj1, event, afb_msg_json_event(event, object)); } +static int wsreq_subscribe(struct afb_wsreq *wsreq, struct afb_event event) +{ + return afb_evt_add_watch(wsreq->aws->listener, event); +} + +static int wsreq_unsubscribe(struct afb_wsreq *wsreq, struct afb_event event) +{ + return afb_evt_remove_watch(wsreq->aws->listener, event); +} + diff --git a/src/afb-ws-json1.h b/src/afb-ws-json1.h index fedbcf6c..f714c222 100644 --- a/src/afb-ws-json1.h +++ b/src/afb-ws-json1.h @@ -19,6 +19,9 @@ struct afb_ws_json1; struct afb_context; +struct afb_req_itf; + +extern const struct afb_req_itf afb_ws_json1_req_itf; extern struct afb_ws_json1 *afb_ws_json1_create(int fd, struct afb_context *context, void (*cleanup)(void*), void *closure); diff --git a/src/session.c b/src/session.c index 16dc8369..e0d0a8e6 100644 --- a/src/session.c +++ b/src/session.c @@ -38,13 +38,6 @@ struct client_value void (*free_value)(void*); }; -struct afb_event_listener_list -{ - struct afb_event_listener_list *next; - struct afb_event_listener listener; - int refcount; -}; - struct AFB_clientCtx { unsigned refcount; @@ -54,7 +47,6 @@ struct AFB_clientCtx char uuid[37]; // long term authentication of remote client char token[37]; // short term authentication of remote client struct client_value *values; - struct afb_event_listener_list *listeners; }; // Session UUID are store in a simple array [for 10 sessions this should be enough] @@ -66,7 +58,6 @@ static struct { int timeout; int apicount; char initok[37]; - struct afb_event_listener_list *listeners; } sessions; /* generate a uuid */ @@ -289,8 +280,6 @@ void ctxClientClose (struct AFB_clientCtx *clientCtx) if (clientCtx->uuid[0] != 0) { clientCtx->uuid[0] = 0; ctxUuidFreeCB (clientCtx); - while(clientCtx->listeners != NULL) - ctxClientEventListenerRemove(clientCtx, clientCtx->listeners->listener); if (clientCtx->refcount == 0) { ctxStoreDel (clientCtx); free(clientCtx); @@ -326,103 +315,6 @@ void ctxTokenNew (struct AFB_clientCtx *clientCtx) clientCtx->expiration = NOW + sessions.timeout; } -static int add_listener(struct afb_event_listener_list **head, struct afb_event_listener listener) -{ - struct afb_event_listener_list *iter, **prv; - - prv = head; - for (;;) { - iter = *prv; - if (iter == NULL) { - iter = calloc(1, sizeof *iter); - if (iter == NULL) { - errno = ENOMEM; - return -1; - } - iter->listener = listener; - iter->refcount = 1; - *prv = iter; - return 0; - } - if (iter->listener.itf == listener.itf && iter->listener.closure == listener.closure) { - iter->refcount++; - return 0; - } - prv = &iter->next; - } -} - -int ctxClientEventListenerAdd(struct AFB_clientCtx *clientCtx, struct afb_event_listener listener) -{ - return add_listener(clientCtx != NULL ? &clientCtx->listeners : &sessions.listeners, listener); -} - -static void remove_listener(struct afb_event_listener_list **head, struct afb_event_listener listener) -{ - struct afb_event_listener_list *iter, **prv; - - prv = head; - for (;;) { - iter = *prv; - if (iter == NULL) - return; - if (iter->listener.itf == listener.itf && iter->listener.closure == listener.closure) { - if (!--iter->refcount) { - *prv = iter->next; - free(iter); - } - return; - } - prv = &iter->next; - } -} - -void ctxClientEventListenerRemove(struct AFB_clientCtx *clientCtx, struct afb_event_listener listener) -{ - remove_listener(clientCtx != NULL ? &clientCtx->listeners : &sessions.listeners, listener); -} - -static int send(struct afb_event_listener_list *head, const char *event, struct json_object *object) -{ - struct afb_event_listener_list *iter; - int result; - - result = 0; - iter = head; - while (iter != NULL) { - if (iter->listener.itf->expects == NULL || iter->listener.itf->expects(iter->listener.closure, event)) { - iter->listener.itf->send(iter->listener.closure, event, json_object_get(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; - - now = NOW; - if (clientCtx != NULL) { - result = ctxIsActive(clientCtx, now) ? send(clientCtx->listeners, event, object) : 0; - } else { - result = send(sessions.listeners, event, object); - for (idx=0; idx < sessions.max; idx++) { - clientCtx = ctxClientAddRef(sessions.store[idx]); - if (clientCtx != NULL && ctxIsActive(clientCtx, now)) { - clientCtx = ctxClientAddRef(clientCtx); - result += send(clientCtx->listeners, event, object); - } - ctxClientUnref(clientCtx); - } - } - return result; -} - const char *ctxClientGetUuid (struct AFB_clientCtx *clientCtx) { assert(clientCtx != NULL); diff --git a/src/session.h b/src/session.h index af074100..497951af 100644 --- a/src/session.h +++ b/src/session.h @@ -20,18 +20,6 @@ struct json_object; struct AFB_clientCtx; -struct afb_event_listener_itf -{ - void (*send)(void *closure, const char *event, struct json_object *object); - int (*expects)(void *closure, const char *event); -}; - -struct afb_event_listener -{ - const struct afb_event_listener_itf *itf; - void *closure; -}; - extern void ctxStoreInit (int max_session_count, int timeout, const char *initok, int context_count); extern struct AFB_clientCtx *ctxClientGetSession (const char *uuid, int *created); @@ -39,10 +27,6 @@ extern struct AFB_clientCtx *ctxClientAddRef(struct AFB_clientCtx *clientCtx); extern void ctxClientUnref(struct AFB_clientCtx *clientCtx); extern void ctxClientClose (struct AFB_clientCtx *clientCtx); -extern int ctxClientEventListenerAdd(struct AFB_clientCtx *clientCtx, struct afb_event_listener listener); -extern void ctxClientEventListenerRemove(struct AFB_clientCtx *clientCtx, struct afb_event_listener listener); -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 void ctxTokenNew (struct AFB_clientCtx *clientCtx); |