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>
35 * Structure for event listeners
37 struct afb_evt_listener {
39 /* chaining listeners */
40 struct afb_evt_listener *next;
42 /* interface for callbacks */
43 const struct afb_evt_itf *itf;
45 /* closure for the callback */
48 /* head of the list of events listened */
49 struct afb_evt_watch *watchs;
51 /* mutex of the listener */
52 pthread_mutex_t mutex;
54 /* count of reference to the listener */
59 * Structure for describing events
61 struct afb_evt_event {
64 struct afb_evt_event *next;
66 /* head of the list of listeners watching the event */
67 struct afb_evt_watch *watchs;
72 /* mutex of the event */
73 pthread_mutex_t mutex;
75 /* name of the event */
80 * Structure for associating events and listeners
82 struct afb_evt_watch {
85 struct afb_evt_event *event;
87 /* link to the next listener for the same event */
88 struct afb_evt_watch *next_by_event;
91 struct afb_evt_listener *listener;
93 /* link to the next event for the same listener */
94 struct afb_evt_watch *next_by_listener;
100 /* declare functions */
101 static int evt_broadcast(struct afb_evt_event *evt, struct json_object *obj);
102 static int evt_push(struct afb_evt_event *evt, struct json_object *obj);
103 static void evt_destroy(struct afb_evt_event *evt);
104 static const char *evt_name(struct afb_evt_event *evt);
106 /* the interface for events */
107 static struct afb_event_itf afb_evt_event_itf = {
108 .broadcast = (void*)evt_broadcast,
109 .push = (void*)evt_push,
110 .drop = (void*)evt_destroy,
111 .name = (void*)evt_name
114 /* head of the list of listeners */
115 static pthread_mutex_t listeners_mutex = PTHREAD_MUTEX_INITIALIZER;
116 static struct afb_evt_listener *listeners = NULL;
118 /* handling id of events */
119 static pthread_mutex_t events_mutex = PTHREAD_MUTEX_INITIALIZER;
120 static struct afb_evt_event *events = NULL;
121 static int event_id_counter = 0;
122 static int event_id_wrapped = 0;
125 * Broadcasts the event 'evt' with its 'object'
126 * 'object' is released (like json_object_put)
127 * Returns the count of listener that received the event.
129 static int evt_broadcast(struct afb_evt_event *evt, struct json_object *object)
131 return afb_evt_broadcast(evt->name, object);
135 * Broadcasts the 'event' with its 'object'
136 * 'object' is released (like json_object_put)
137 * Returns the count of listener having receive the event.
139 int afb_evt_broadcast(const char *event, struct json_object *object)
142 struct afb_evt_listener *listener;
145 pthread_mutex_lock(&listeners_mutex);
146 listener = listeners;
148 if (listener->itf->broadcast != NULL) {
149 listener->itf->broadcast(listener->closure, event, 0, json_object_get(object));
152 listener = listener->next;
154 pthread_mutex_unlock(&listeners_mutex);
155 json_object_put(object);
160 * Pushes the event 'evt' with 'obj' to its listeners
161 * 'obj' is released (like json_object_put)
162 * Returns the count of listener taht received the event.
164 static int evt_push(struct afb_evt_event *evt, struct json_object *obj)
167 struct afb_evt_watch *watch;
168 struct afb_evt_listener *listener;
171 pthread_mutex_lock(&evt->mutex);
174 listener = watch->listener;
175 assert(listener->itf->push != NULL);
176 if (watch->activity != 0)
177 listener->itf->push(listener->closure, evt->name, evt->id, json_object_get(obj));
178 watch = watch->next_by_event;
181 pthread_mutex_unlock(&evt->mutex);
182 json_object_put(obj);
187 * Returns the name associated to the event 'evt'.
189 static const char *evt_name(struct afb_evt_event *evt)
197 static void remove_watch(struct afb_evt_watch *watch)
199 struct afb_evt_watch **prv;
200 struct afb_evt_event *evt;
201 struct afb_evt_listener *listener;
203 /* notify listener if needed */
205 listener = watch->listener;
206 if (watch->activity != 0 && listener->itf->remove != NULL)
207 listener->itf->remove(listener->closure, evt->name, evt->id);
209 /* unlink the watch for its event */
212 prv = &(*prv)->next_by_event;
213 *prv = watch->next_by_event;
215 /* unlink the watch for its listener */
216 prv = &listener->watchs;
218 prv = &(*prv)->next_by_listener;
219 *prv = watch->next_by_listener;
226 * Destroys the event 'evt'
228 static void evt_destroy(struct afb_evt_event *evt)
231 struct afb_evt_event **prv;
232 struct afb_evt_listener *listener;
235 /* unlinks the event if valid! */
236 pthread_mutex_lock(&events_mutex);
239 while (*prv && !(found = (*prv == evt)))
243 pthread_mutex_unlock(&events_mutex);
245 /* destroys the event */
247 /* removes all watchers */
248 while(evt->watchs != NULL) {
249 listener = evt->watchs->listener;
250 pthread_mutex_lock(&listener->mutex);
251 pthread_mutex_lock(&evt->mutex);
252 remove_watch(evt->watchs);
253 pthread_mutex_unlock(&evt->mutex);
254 pthread_mutex_unlock(&listener->mutex);
258 pthread_mutex_destroy(&evt->mutex);
265 * Creates an event of 'name' and returns it.
266 * Returns an event with closure==NULL in case of error.
268 struct afb_event afb_evt_create_event(const char *name)
271 struct afb_evt_event *evt;
273 /* allocates the event */
275 evt = malloc(len + sizeof * evt);
279 /* initialize the event */
281 memcpy(evt->name, name, len + 1);
283 /* allocates the id */
284 pthread_mutex_lock(&events_mutex);
286 if (++event_id_counter < 0) {
287 event_id_wrapped = 1;
288 event_id_counter = 1024; /* heuristic: small numbers are not destroyed */
290 if (!event_id_wrapped)
293 while(evt != NULL && evt->id != event_id_counter)
295 } while (evt != NULL);
297 /* initialize the event */
298 memcpy(evt->name, name, len + 1);
301 evt->id = event_id_counter;
302 pthread_mutex_init(&evt->mutex, NULL);
304 pthread_mutex_unlock(&events_mutex);
306 /* returns the event */
307 return (struct afb_event){ .itf = &afb_evt_event_itf, .closure = evt };
309 return (struct afb_event){ .itf = NULL, .closure = NULL };
313 * Returns the name of the 'event'
315 const char *afb_evt_event_name(struct afb_event event)
317 return (event.itf != &afb_evt_event_itf) ? NULL : ((struct afb_evt_event *)event.closure)->name;
321 * Returns the id of the 'event'
323 int afb_evt_event_id(struct afb_event event)
325 return (event.itf != &afb_evt_event_itf) ? 0 : ((struct afb_evt_event *)event.closure)->id;
329 * Returns an instance of the listener defined by the 'send' callback
331 * Returns NULL in case of memory depletion.
333 struct afb_evt_listener *afb_evt_listener_create(const struct afb_evt_itf *itf, void *closure)
335 struct afb_evt_listener *listener;
337 /* search if an instance already exists */
338 pthread_mutex_lock(&listeners_mutex);
339 listener = listeners;
340 while (listener != NULL) {
341 if (listener->itf == itf && listener->closure == closure) {
342 listener = afb_evt_listener_addref(listener);
345 listener = listener->next;
349 listener = calloc(1, sizeof *listener);
350 if (listener != NULL) {
353 listener->closure = closure;
354 listener->watchs = NULL;
355 listener->refcount = 1;
356 pthread_mutex_init(&listener->mutex, NULL);
357 listener->next = listeners;
358 listeners = listener;
361 pthread_mutex_unlock(&listeners_mutex);
366 * Increases the reference count of 'listener' and returns it
368 struct afb_evt_listener *afb_evt_listener_addref(struct afb_evt_listener *listener)
370 __atomic_add_fetch(&listener->refcount, 1, __ATOMIC_RELAXED);
375 * Decreases the reference count of the 'listener' and destroys it
378 void afb_evt_listener_unref(struct afb_evt_listener *listener)
380 struct afb_evt_listener **prv;
381 struct afb_evt_event *evt;
383 if (!__atomic_sub_fetch(&listener->refcount, 1, __ATOMIC_RELAXED)) {
385 /* unlink the listener */
386 pthread_mutex_lock(&listeners_mutex);
388 while (*prv != listener)
390 *prv = listener->next;
391 pthread_mutex_unlock(&listeners_mutex);
393 /* remove the watchers */
394 pthread_mutex_lock(&listener->mutex);
395 while (listener->watchs != NULL) {
396 evt = listener->watchs->event;
397 pthread_mutex_lock(&evt->mutex);
398 remove_watch(listener->watchs);
399 pthread_mutex_unlock(&evt->mutex);
401 pthread_mutex_unlock(&listener->mutex);
403 /* free the listener */
404 pthread_mutex_destroy(&listener->mutex);
410 * Makes the 'listener' watching 'event'
411 * Returns 0 in case of success or else -1.
413 int afb_evt_add_watch(struct afb_evt_listener *listener, struct afb_event event)
415 struct afb_evt_watch *watch;
416 struct afb_evt_event *evt;
418 /* check parameter */
419 if (event.itf != &afb_evt_event_itf || listener->itf->push == NULL) {
424 /* search the existing watch for the listener */
426 pthread_mutex_lock(&listener->mutex);
427 watch = listener->watchs;
428 while(watch != NULL) {
429 if (watch->event == evt)
431 watch = watch->next_by_listener;
434 /* not found, allocate a new */
435 watch = malloc(sizeof *watch);
437 pthread_mutex_unlock(&listener->mutex);
442 /* initialise and link */
445 watch->listener = listener;
446 watch->next_by_listener = listener->watchs;
447 listener->watchs = watch;
448 pthread_mutex_lock(&evt->mutex);
449 watch->next_by_event = evt->watchs;
451 pthread_mutex_unlock(&evt->mutex);
454 if (watch->activity == 0 && listener->itf->add != NULL)
455 listener->itf->add(listener->closure, evt->name, evt->id);
457 pthread_mutex_unlock(&listener->mutex);
463 * Avoids the 'listener' to watch 'event'
464 * Returns 0 in case of success or else -1.
466 int afb_evt_remove_watch(struct afb_evt_listener *listener, struct afb_event event)
468 struct afb_evt_watch *watch;
469 struct afb_evt_event *evt;
471 /* check parameter */
472 if (event.itf != &afb_evt_event_itf) {
477 /* search the existing watch */
479 pthread_mutex_lock(&listener->mutex);
480 watch = listener->watchs;
481 while(watch != NULL) {
482 if (watch->event == evt) {
483 /* found: remove it */
484 if (watch->activity != 0) {
486 if (watch->activity == 0 && listener->itf->remove != NULL)
487 listener->itf->remove(listener->closure, evt->name, evt->id);
489 pthread_mutex_unlock(&listener->mutex);
492 watch = watch->next_by_listener;
494 pthread_mutex_unlock(&listener->mutex);