2 * Copyright (C) 2015, 2016, 2017 "IoT.bzh"
3 * Author "Fulup Ar Foll"
4 * Author José Bollo <jose.bollo@iot.bzh>
6 * Licensed under the Apache License, Version 2.0 (the "License");
7 * you may not use this file except in compliance with the License.
8 * You may obtain a copy of the License at
10 * http://www.apache.org/licenses/LICENSE-2.0
12 * Unless required by applicable law or agreed to in writing, software
13 * distributed under the License is distributed on an "AS IS" BASIS,
14 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15 * See the License for the specific language governing permissions and
16 * limitations under the License.
27 #include <json-c/json.h>
28 #include <afb/afb-event-itf.h>
37 * Structure for event listeners
39 struct afb_evt_listener {
41 /* chaining listeners */
42 struct afb_evt_listener *next;
44 /* interface for callbacks */
45 const struct afb_evt_itf *itf;
47 /* closure for the callback */
50 /* head of the list of events listened */
51 struct afb_evt_watch *watchs;
53 /* mutex of the listener */
54 pthread_mutex_t mutex;
56 /* count of reference to the listener */
61 * Structure for describing events
63 struct afb_evt_event {
66 struct afb_evt_event *next;
68 /* head of the list of listeners watching the event */
69 struct afb_evt_watch *watchs;
77 /* mutex of the event */
78 pthread_mutex_t mutex;
80 /* name of the event */
85 * Structure for associating events and listeners
87 struct afb_evt_watch {
90 struct afb_evt_event *event;
92 /* link to the next listener for the same event */
93 struct afb_evt_watch *next_by_event;
96 struct afb_evt_listener *listener;
98 /* link to the next event for the same listener */
99 struct afb_evt_watch *next_by_listener;
105 /* declare functions */
106 static int evt_broadcast(struct afb_evt_event *evt, struct json_object *obj);
107 static int evt_push(struct afb_evt_event *evt, struct json_object *obj);
108 static void evt_destroy(struct afb_evt_event *evt);
109 static const char *evt_name(struct afb_evt_event *evt);
111 /* the interface for events */
112 static struct afb_event_itf afb_evt_event_itf = {
113 .broadcast = (void*)evt_broadcast,
114 .push = (void*)evt_push,
115 .drop = (void*)evt_destroy,
116 .name = (void*)evt_name
119 /* head of the list of listeners */
120 static pthread_mutex_t listeners_mutex = PTHREAD_MUTEX_INITIALIZER;
121 static struct afb_evt_listener *listeners = NULL;
123 /* handling id of events */
124 static pthread_mutex_t events_mutex = PTHREAD_MUTEX_INITIALIZER;
125 static struct afb_evt_event *events = NULL;
126 static int event_id_counter = 0;
127 static int event_id_wrapped = 0;
130 * Broadcasts the 'event' of 'id' with its 'obj'
131 * 'obj' is released (like json_object_put)
132 * calls hooks if hookflags isn't 0
133 * Returns the count of listener having receive the event.
135 static int broadcast(const char *event, struct json_object *obj, int id, int hookflags)
138 struct afb_evt_listener *listener;
140 if (hookflags & afb_hook_flag_evt_broadcast_before)
141 afb_hook_evt_broadcast_before(event, id, obj);
143 pthread_mutex_lock(&listeners_mutex);
144 listener = listeners;
146 if (listener->itf->broadcast != NULL) {
147 listener->itf->broadcast(listener->closure, event, id, json_object_get(obj));
150 listener = listener->next;
152 pthread_mutex_unlock(&listeners_mutex);
153 if (hookflags & afb_hook_flag_evt_broadcast_after)
154 afb_hook_evt_broadcast_after(event, id, obj, result);
155 json_object_put(obj);
160 * Broadcasts the event 'evt' with its 'object'
161 * 'object' is released (like json_object_put)
162 * Returns the count of listener that received the event.
164 static int evt_broadcast(struct afb_evt_event *evt, struct json_object *object)
166 return broadcast(evt->name, object, evt->id, evt->hookflags);
170 * Broadcasts the 'event' with its 'object'
171 * 'object' is released (like json_object_put)
172 * Returns the count of listener having receive the event.
174 int afb_evt_broadcast(const char *event, struct json_object *object)
176 return broadcast(event, object, 0, -1);
180 * Pushes the event 'evt' with 'obj' to its listeners
181 * 'obj' is released (like json_object_put)
182 * calls hooks if hookflags isn't 0
183 * Returns the count of listener taht received the event.
185 static int push(struct afb_evt_event *evt, struct json_object *obj, int hookflags)
188 struct afb_evt_watch *watch;
189 struct afb_evt_listener *listener;
192 pthread_mutex_lock(&evt->mutex);
193 if (hookflags & afb_hook_flag_evt_push_before)
194 afb_hook_evt_push_before(evt->name, evt->id, obj);
197 listener = watch->listener;
198 assert(listener->itf->push != NULL);
199 if (watch->activity != 0) {
200 listener->itf->push(listener->closure, evt->name, evt->id, json_object_get(obj));
203 watch = watch->next_by_event;
205 if (hookflags & afb_hook_flag_evt_push_after)
206 afb_hook_evt_push_after(evt->name, evt->id, obj, result);
207 pthread_mutex_unlock(&evt->mutex);
208 json_object_put(obj);
213 * Pushes the event 'evt' with 'obj' to its listeners
214 * 'obj' is released (like json_object_put)
215 * Returns the count of listener taht received the event.
217 static int evt_push(struct afb_evt_event *evt, struct json_object *obj)
219 return push(evt, obj, evt->hookflags);
223 * Returns the name associated to the event 'evt'.
225 static const char *evt_name(struct afb_evt_event *evt)
227 const char *name = strchr(evt->name, '/');
228 name = name ? name + 1 : evt->name;
229 if (evt->hookflags & afb_hook_flag_evt_name)
230 afb_hook_evt_name(evt->name, evt->id);
237 static void remove_watch(struct afb_evt_watch *watch)
239 struct afb_evt_watch **prv;
240 struct afb_evt_event *evt;
241 struct afb_evt_listener *listener;
243 /* notify listener if needed */
245 listener = watch->listener;
246 if (watch->activity != 0 && listener->itf->remove != NULL)
247 listener->itf->remove(listener->closure, evt->name, evt->id);
249 /* unlink the watch for its event */
252 prv = &(*prv)->next_by_event;
253 *prv = watch->next_by_event;
255 /* unlink the watch for its listener */
256 prv = &listener->watchs;
258 prv = &(*prv)->next_by_listener;
259 *prv = watch->next_by_listener;
266 * Destroys the event 'evt'
268 static void evt_destroy(struct afb_evt_event *evt)
271 struct afb_evt_event **prv;
272 struct afb_evt_listener *listener;
275 /* unlinks the event if valid! */
276 pthread_mutex_lock(&events_mutex);
279 while (*prv && !(found = (*prv == evt)))
283 pthread_mutex_unlock(&events_mutex);
285 /* destroys the event */
287 /* removes all watchers */
288 while(evt->watchs != NULL) {
289 listener = evt->watchs->listener;
290 pthread_mutex_lock(&listener->mutex);
291 pthread_mutex_lock(&evt->mutex);
292 remove_watch(evt->watchs);
293 pthread_mutex_unlock(&evt->mutex);
294 pthread_mutex_unlock(&listener->mutex);
298 if (evt->hookflags & afb_hook_flag_evt_drop)
299 afb_hook_evt_drop(evt->name, evt->id);
302 pthread_mutex_destroy(&evt->mutex);
309 * Creates an event of 'name' and returns it.
310 * Returns an event with closure==NULL in case of error.
312 struct afb_event afb_evt_create_event(const char *name)
315 struct afb_evt_event *evt;
317 /* allocates the event */
319 evt = malloc(len + sizeof * evt);
323 /* initialize the event */
325 memcpy(evt->name, name, len + 1);
327 /* allocates the id */
328 pthread_mutex_lock(&events_mutex);
330 if (++event_id_counter < 0) {
331 event_id_wrapped = 1;
332 event_id_counter = 1024; /* heuristic: small numbers are not destroyed */
334 if (!event_id_wrapped)
337 while(evt != NULL && evt->id != event_id_counter)
339 } while (evt != NULL);
341 /* initialize the event */
342 memcpy(evt->name, name, len + 1);
345 evt->id = event_id_counter;
346 pthread_mutex_init(&evt->mutex, NULL);
348 evt->hookflags = afb_hook_flags_evt(evt->name);
349 if (evt->hookflags & afb_hook_flag_evt_create)
350 afb_hook_evt_create(evt->name, evt->id);
351 pthread_mutex_unlock(&events_mutex);
353 /* returns the event */
354 return (struct afb_event){ .itf = &afb_evt_event_itf, .closure = evt };
356 return (struct afb_event){ .itf = NULL, .closure = NULL };
360 * Returns the name of the 'event'
362 const char *afb_evt_event_name(struct afb_event event)
364 return (event.itf != &afb_evt_event_itf) ? NULL : ((struct afb_evt_event *)event.closure)->name;
368 * Returns the id of the 'event'
370 int afb_evt_event_id(struct afb_event event)
372 return (event.itf != &afb_evt_event_itf) ? 0 : ((struct afb_evt_event *)event.closure)->id;
376 * Returns an instance of the listener defined by the 'send' callback
378 * Returns NULL in case of memory depletion.
380 struct afb_evt_listener *afb_evt_listener_create(const struct afb_evt_itf *itf, void *closure)
382 struct afb_evt_listener *listener;
384 /* search if an instance already exists */
385 pthread_mutex_lock(&listeners_mutex);
386 listener = listeners;
387 while (listener != NULL) {
388 if (listener->itf == itf && listener->closure == closure) {
389 listener = afb_evt_listener_addref(listener);
392 listener = listener->next;
396 listener = calloc(1, sizeof *listener);
397 if (listener != NULL) {
400 listener->closure = closure;
401 listener->watchs = NULL;
402 listener->refcount = 1;
403 pthread_mutex_init(&listener->mutex, NULL);
404 listener->next = listeners;
405 listeners = listener;
408 pthread_mutex_unlock(&listeners_mutex);
413 * Increases the reference count of 'listener' and returns it
415 struct afb_evt_listener *afb_evt_listener_addref(struct afb_evt_listener *listener)
417 __atomic_add_fetch(&listener->refcount, 1, __ATOMIC_RELAXED);
422 * Decreases the reference count of the 'listener' and destroys it
425 void afb_evt_listener_unref(struct afb_evt_listener *listener)
427 struct afb_evt_listener **prv;
428 struct afb_evt_event *evt;
430 if (!__atomic_sub_fetch(&listener->refcount, 1, __ATOMIC_RELAXED)) {
432 /* unlink the listener */
433 pthread_mutex_lock(&listeners_mutex);
435 while (*prv != listener)
437 *prv = listener->next;
438 pthread_mutex_unlock(&listeners_mutex);
440 /* remove the watchers */
441 pthread_mutex_lock(&listener->mutex);
442 while (listener->watchs != NULL) {
443 evt = listener->watchs->event;
444 pthread_mutex_lock(&evt->mutex);
445 remove_watch(listener->watchs);
446 pthread_mutex_unlock(&evt->mutex);
448 pthread_mutex_unlock(&listener->mutex);
450 /* free the listener */
451 pthread_mutex_destroy(&listener->mutex);
457 * Makes the 'listener' watching 'event'
458 * Returns 0 in case of success or else -1.
460 int afb_evt_add_watch(struct afb_evt_listener *listener, struct afb_event event)
462 struct afb_evt_watch *watch;
463 struct afb_evt_event *evt;
465 /* check parameter */
466 if (event.itf != &afb_evt_event_itf || listener->itf->push == NULL) {
471 /* search the existing watch for the listener */
473 pthread_mutex_lock(&listener->mutex);
474 watch = listener->watchs;
475 while(watch != NULL) {
476 if (watch->event == evt)
478 watch = watch->next_by_listener;
481 /* not found, allocate a new */
482 watch = malloc(sizeof *watch);
484 pthread_mutex_unlock(&listener->mutex);
489 /* initialise and link */
492 watch->listener = listener;
493 watch->next_by_listener = listener->watchs;
494 listener->watchs = watch;
495 pthread_mutex_lock(&evt->mutex);
496 watch->next_by_event = evt->watchs;
498 pthread_mutex_unlock(&evt->mutex);
501 if (watch->activity == 0 && listener->itf->add != NULL)
502 listener->itf->add(listener->closure, evt->name, evt->id);
504 pthread_mutex_unlock(&listener->mutex);
510 * Avoids the 'listener' to watch 'event'
511 * Returns 0 in case of success or else -1.
513 int afb_evt_remove_watch(struct afb_evt_listener *listener, struct afb_event event)
515 struct afb_evt_watch *watch;
516 struct afb_evt_event *evt;
518 /* check parameter */
519 if (event.itf != &afb_evt_event_itf) {
524 /* search the existing watch */
526 pthread_mutex_lock(&listener->mutex);
527 watch = listener->watchs;
528 while(watch != NULL) {
529 if (watch->event == evt) {
530 if (watch->activity != 0) {
532 if (watch->activity == 0 && listener->itf->remove != NULL)
533 listener->itf->remove(listener->closure, evt->name, evt->id);
535 pthread_mutex_unlock(&listener->mutex);
538 watch = watch->next_by_listener;
540 pthread_mutex_unlock(&listener->mutex);
546 * update the hooks for events
548 void afb_evt_update_hooks()
550 struct afb_evt_event *evt;
552 pthread_mutex_lock(&events_mutex);
553 for (evt = events ; evt ; evt = evt->next)
554 evt->hookflags = afb_hook_flags_evt(evt->name);
555 pthread_mutex_unlock(&events_mutex);
558 int afb_evt_push(struct afb_event event, struct json_object *object)
560 if (event.itf == &afb_evt_event_itf)
561 return evt_push((struct afb_evt_event *)event.closure, object);
562 json_object_put(object);
566 int afb_evt_unhooked_push(struct afb_event event, struct json_object *object)
568 if (event.itf == &afb_evt_event_itf)
569 return push((struct afb_evt_event *)event.closure, object, 0);
570 json_object_put(object);