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>
36 * Structure for event listeners
38 struct afb_evt_listener {
40 /* chaining listeners */
41 struct afb_evt_listener *next;
43 /* interface for callbacks */
44 const struct afb_evt_itf *itf;
46 /* closure for the callback */
49 /* head of the list of events listened */
50 struct afb_evt_watch *watchs;
52 /* mutex of the listener */
53 pthread_mutex_t mutex;
55 /* count of reference to the listener */
60 * Structure for describing events
62 struct afb_evt_event {
65 struct afb_evt_event *next;
67 /* head of the list of listeners watching the event */
68 struct afb_evt_watch *watchs;
73 /* mutex of the event */
74 pthread_mutex_t mutex;
76 /* name of the event */
81 * Structure for associating events and listeners
83 struct afb_evt_watch {
86 struct afb_evt_event *event;
88 /* link to the next listener for the same event */
89 struct afb_evt_watch *next_by_event;
92 struct afb_evt_listener *listener;
94 /* link to the next event for the same listener */
95 struct afb_evt_watch *next_by_listener;
101 /* declare functions */
102 static int evt_broadcast(struct afb_evt_event *evt, struct json_object *obj);
103 static int evt_push(struct afb_evt_event *evt, struct json_object *obj);
104 static void evt_destroy(struct afb_evt_event *evt);
105 static const char *evt_name(struct afb_evt_event *evt);
107 /* the interface for events */
108 static struct afb_event_itf afb_evt_event_itf = {
109 .broadcast = (void*)evt_broadcast,
110 .push = (void*)evt_push,
111 .drop = (void*)evt_destroy,
112 .name = (void*)evt_name
115 /* head of the list of listeners */
116 static pthread_mutex_t listeners_mutex = PTHREAD_MUTEX_INITIALIZER;
117 static struct afb_evt_listener *listeners = NULL;
119 /* handling id of events */
120 static pthread_mutex_t events_mutex = PTHREAD_MUTEX_INITIALIZER;
121 static struct afb_evt_event *events = NULL;
122 static int event_id_counter = 0;
123 static int event_id_wrapped = 0;
126 * Broadcasts the event 'evt' with its 'object'
127 * 'object' is released (like json_object_put)
128 * Returns the count of listener that received the event.
130 static int evt_broadcast(struct afb_evt_event *evt, struct json_object *object)
132 return afb_evt_broadcast(evt->name, object);
136 * Broadcasts the 'event' with its 'object'
137 * 'object' is released (like json_object_put)
138 * Returns the count of listener having receive the event.
140 int afb_evt_broadcast(const char *event, struct json_object *object)
143 struct afb_evt_listener *listener;
146 pthread_mutex_lock(&listeners_mutex);
147 listener = listeners;
149 if (listener->itf->broadcast != NULL) {
150 listener->itf->broadcast(listener->closure, event, 0, json_object_get(object));
153 listener = listener->next;
155 pthread_mutex_unlock(&listeners_mutex);
156 json_object_put(object);
161 * Pushes the event 'evt' with 'obj' to its listeners
162 * 'obj' is released (like json_object_put)
163 * Returns the count of listener taht received the event.
165 static int evt_push(struct afb_evt_event *evt, struct json_object *obj)
168 struct afb_evt_watch *watch;
169 struct afb_evt_listener *listener;
172 pthread_mutex_lock(&evt->mutex);
175 listener = watch->listener;
176 assert(listener->itf->push != NULL);
177 if (watch->activity != 0) {
178 listener->itf->push(listener->closure, evt->name, evt->id, json_object_get(obj));
181 watch = watch->next_by_event;
183 pthread_mutex_unlock(&evt->mutex);
184 json_object_put(obj);
189 * Returns the name associated to the event 'evt'.
191 static const char *evt_name(struct afb_evt_event *evt)
199 static void remove_watch(struct afb_evt_watch *watch)
201 struct afb_evt_watch **prv;
202 struct afb_evt_event *evt;
203 struct afb_evt_listener *listener;
205 /* notify listener if needed */
207 listener = watch->listener;
208 if (watch->activity != 0 && listener->itf->remove != NULL)
209 listener->itf->remove(listener->closure, evt->name, evt->id);
211 /* unlink the watch for its event */
214 prv = &(*prv)->next_by_event;
215 *prv = watch->next_by_event;
217 /* unlink the watch for its listener */
218 prv = &listener->watchs;
220 prv = &(*prv)->next_by_listener;
221 *prv = watch->next_by_listener;
228 * Destroys the event 'evt'
230 static void evt_destroy(struct afb_evt_event *evt)
233 struct afb_evt_event **prv;
234 struct afb_evt_listener *listener;
237 /* unlinks the event if valid! */
238 pthread_mutex_lock(&events_mutex);
241 while (*prv && !(found = (*prv == evt)))
245 pthread_mutex_unlock(&events_mutex);
247 /* destroys the event */
249 /* removes all watchers */
250 while(evt->watchs != NULL) {
251 listener = evt->watchs->listener;
252 pthread_mutex_lock(&listener->mutex);
253 pthread_mutex_lock(&evt->mutex);
254 remove_watch(evt->watchs);
255 pthread_mutex_unlock(&evt->mutex);
256 pthread_mutex_unlock(&listener->mutex);
260 pthread_mutex_destroy(&evt->mutex);
267 * Creates an event of 'name' and returns it.
268 * Returns an event with closure==NULL in case of error.
270 struct afb_event afb_evt_create_event(const char *name)
273 struct afb_evt_event *evt;
275 /* allocates the event */
277 evt = malloc(len + sizeof * evt);
281 /* initialize the event */
283 memcpy(evt->name, name, len + 1);
285 /* allocates the id */
286 pthread_mutex_lock(&events_mutex);
288 if (++event_id_counter < 0) {
289 event_id_wrapped = 1;
290 event_id_counter = 1024; /* heuristic: small numbers are not destroyed */
292 if (!event_id_wrapped)
295 while(evt != NULL && evt->id != event_id_counter)
297 } while (evt != NULL);
299 /* initialize the event */
300 memcpy(evt->name, name, len + 1);
303 evt->id = event_id_counter;
304 pthread_mutex_init(&evt->mutex, NULL);
306 pthread_mutex_unlock(&events_mutex);
308 /* returns the event */
309 return (struct afb_event){ .itf = &afb_evt_event_itf, .closure = evt };
311 return (struct afb_event){ .itf = NULL, .closure = NULL };
315 * Returns the name of the 'event'
317 const char *afb_evt_event_name(struct afb_event event)
319 return (event.itf != &afb_evt_event_itf) ? NULL : ((struct afb_evt_event *)event.closure)->name;
323 * Returns the id of the 'event'
325 int afb_evt_event_id(struct afb_event event)
327 return (event.itf != &afb_evt_event_itf) ? 0 : ((struct afb_evt_event *)event.closure)->id;
331 * Returns an instance of the listener defined by the 'send' callback
333 * Returns NULL in case of memory depletion.
335 struct afb_evt_listener *afb_evt_listener_create(const struct afb_evt_itf *itf, void *closure)
337 struct afb_evt_listener *listener;
339 /* search if an instance already exists */
340 pthread_mutex_lock(&listeners_mutex);
341 listener = listeners;
342 while (listener != NULL) {
343 if (listener->itf == itf && listener->closure == closure) {
344 listener = afb_evt_listener_addref(listener);
347 listener = listener->next;
351 listener = calloc(1, sizeof *listener);
352 if (listener != NULL) {
355 listener->closure = closure;
356 listener->watchs = NULL;
357 listener->refcount = 1;
358 pthread_mutex_init(&listener->mutex, NULL);
359 listener->next = listeners;
360 listeners = listener;
363 pthread_mutex_unlock(&listeners_mutex);
368 * Increases the reference count of 'listener' and returns it
370 struct afb_evt_listener *afb_evt_listener_addref(struct afb_evt_listener *listener)
372 __atomic_add_fetch(&listener->refcount, 1, __ATOMIC_RELAXED);
377 * Decreases the reference count of the 'listener' and destroys it
380 void afb_evt_listener_unref(struct afb_evt_listener *listener)
382 struct afb_evt_listener **prv;
383 struct afb_evt_event *evt;
385 if (!__atomic_sub_fetch(&listener->refcount, 1, __ATOMIC_RELAXED)) {
387 /* unlink the listener */
388 pthread_mutex_lock(&listeners_mutex);
390 while (*prv != listener)
392 *prv = listener->next;
393 pthread_mutex_unlock(&listeners_mutex);
395 /* remove the watchers */
396 pthread_mutex_lock(&listener->mutex);
397 while (listener->watchs != NULL) {
398 evt = listener->watchs->event;
399 pthread_mutex_lock(&evt->mutex);
400 remove_watch(listener->watchs);
401 pthread_mutex_unlock(&evt->mutex);
403 pthread_mutex_unlock(&listener->mutex);
405 /* free the listener */
406 pthread_mutex_destroy(&listener->mutex);
412 * Makes the 'listener' watching 'event'
413 * Returns 0 in case of success or else -1.
415 int afb_evt_add_watch(struct afb_evt_listener *listener, struct afb_event event)
417 struct afb_evt_watch *watch;
418 struct afb_evt_event *evt;
420 /* check parameter */
421 if (event.itf != &afb_evt_event_itf || listener->itf->push == NULL) {
426 /* search the existing watch for the listener */
428 pthread_mutex_lock(&listener->mutex);
429 watch = listener->watchs;
430 while(watch != NULL) {
431 if (watch->event == evt)
433 watch = watch->next_by_listener;
436 /* not found, allocate a new */
437 watch = malloc(sizeof *watch);
439 pthread_mutex_unlock(&listener->mutex);
444 /* initialise and link */
447 watch->listener = listener;
448 watch->next_by_listener = listener->watchs;
449 listener->watchs = watch;
450 pthread_mutex_lock(&evt->mutex);
451 watch->next_by_event = evt->watchs;
453 pthread_mutex_unlock(&evt->mutex);
456 if (watch->activity == 0 && listener->itf->add != NULL)
457 listener->itf->add(listener->closure, evt->name, evt->id);
459 pthread_mutex_unlock(&listener->mutex);
465 * Avoids the 'listener' to watch 'event'
466 * Returns 0 in case of success or else -1.
468 int afb_evt_remove_watch(struct afb_evt_listener *listener, struct afb_event event)
470 struct afb_evt_watch *watch;
471 struct afb_evt_event *evt;
473 /* check parameter */
474 if (event.itf != &afb_evt_event_itf) {
479 /* search the existing watch */
481 pthread_mutex_lock(&listener->mutex);
482 watch = listener->watchs;
483 while(watch != NULL) {
484 if (watch->event == evt) {
485 if (watch->activity != 0) {
487 if (watch->activity == 0 && listener->itf->remove != NULL)
488 listener->itf->remove(listener->closure, evt->name, evt->id);
490 pthread_mutex_unlock(&listener->mutex);
493 watch = watch->next_by_listener;
495 pthread_mutex_unlock(&listener->mutex);