2 * Copyright (C) 2015-2018 "IoT.bzh"
3 * Author José Bollo <jose.bollo@iot.bzh>
5 * Licensed under the Apache License, Version 2.0 (the "License");
6 * you may not use this file except in compliance with the License.
7 * You may obtain a copy of the License at
9 * http://www.apache.org/licenses/LICENSE-2.0
11 * Unless required by applicable law or agreed to in writing, software
12 * distributed under the License is distributed on an "AS IS" BASIS,
13 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 * See the License for the specific language governing permissions and
15 * limitations under the License.
26 #include <json-c/json.h>
27 #include <afb/afb-event-x2-itf.h>
28 #include <afb/afb-event-x1.h>
38 * Structure for event listeners
40 struct afb_evt_listener {
42 /* chaining listeners */
43 struct afb_evt_listener *next;
45 /* interface for callbacks */
46 const struct afb_evt_itf *itf;
48 /* closure for the callback */
51 /* head of the list of events listened */
52 struct afb_evt_watch *watchs;
54 /* rwlock of the listener */
55 pthread_rwlock_t rwlock;
57 /* count of reference to the listener */
62 * Structure for describing events
67 struct afb_event_x2 eventid;
70 struct afb_evtid *next;
72 /* head of the list of listeners watching the event */
73 struct afb_evt_watch *watchs;
75 /* rwlock of the event */
76 pthread_rwlock_t rwlock;
90 /* fullname of the event */
95 * Structure for associating events and listeners
97 struct afb_evt_watch {
100 struct afb_evtid *evtid;
102 /* link to the next watcher for the same evtid */
103 struct afb_evt_watch *next_by_evtid;
106 struct afb_evt_listener *listener;
108 /* link to the next watcher for the same listener */
109 struct afb_evt_watch *next_by_listener;
116 * structure for job of broadcasting events
120 /** object atached to the event */
121 struct json_object *object;
123 /** name of the event to broadcast */
128 * structure for job of broadcasting or pushing events
132 /** the event to broadcast */
133 struct afb_evtid *evtid;
135 /** object atached to the event */
136 struct json_object *object;
139 /* the interface for events */
140 static struct afb_event_x2_itf afb_evt_event_x2_itf = {
141 .broadcast = (void*)afb_evt_evtid_broadcast,
142 .push = (void*)afb_evt_evtid_push,
143 .unref = (void*)afb_evt_evtid_unref,
144 .name = (void*)afb_evt_evtid_name,
145 .addref = (void*)afb_evt_evtid_addref
148 /* the interface for events */
149 static struct afb_event_x2_itf afb_evt_hooked_eventid_itf = {
150 .broadcast = (void*)afb_evt_evtid_hooked_broadcast,
151 .push = (void*)afb_evt_evtid_hooked_push,
152 .unref = (void*)afb_evt_evtid_hooked_unref,
153 .name = (void*)afb_evt_evtid_hooked_name,
154 .addref = (void*)afb_evt_evtid_hooked_addref
157 /* job groups for events push/broadcast */
158 #define BROADCAST_JOB_GROUP (&afb_evt_event_x2_itf)
159 #define PUSH_JOB_GROUP (&afb_evt_event_x2_itf)
161 /* head of the list of listeners */
162 static pthread_rwlock_t listeners_rwlock = PTHREAD_RWLOCK_INITIALIZER;
163 static struct afb_evt_listener *listeners = NULL;
165 /* handling id of events */
166 static pthread_rwlock_t events_rwlock = PTHREAD_RWLOCK_INITIALIZER;
167 static struct afb_evtid *evtids = NULL;
168 static int event_id_counter = 0;
169 static int event_id_wrapped = 0;
172 * Create structure for job of broadcasting string 'event' with 'object'
173 * Returns the created structure or NULL if out of memory
175 static struct job_broadcast *make_job_broadcast(const char *event, struct json_object *object)
177 size_t sz = 1 + strlen(event);
178 struct job_broadcast *jb = malloc(sz + sizeof *jb);
181 memcpy(jb->event, event, sz);
187 * Destroy structure 'jb' for job of broadcasting string events
189 static void destroy_job_broadcast(struct job_broadcast *jb)
191 json_object_put(jb->object);
196 * Create structure for job of broadcasting or pushing 'evtid' with 'object'
197 * Returns the created structure or NULL if out of memory
199 static struct job_evtid *make_job_evtid(struct afb_evtid *evtid, struct json_object *object)
201 struct job_evtid *je = malloc(sizeof *je);
203 je->evtid = afb_evt_evtid_addref(evtid);
210 * Destroy structure for job of broadcasting or pushing evtid
212 static void destroy_job_evtid(struct job_evtid *je)
214 afb_evt_evtid_unref(je->evtid);
215 json_object_put(je->object);
220 * Broadcasts the 'event' of 'id' with its 'object'
222 static void broadcast(const char *event, struct json_object *object)
224 struct afb_evt_listener *listener;
226 pthread_rwlock_rdlock(&listeners_rwlock);
227 listener = listeners;
229 if (listener->itf->broadcast != NULL)
230 listener->itf->broadcast(listener->closure, event, json_object_get(object));
231 listener = listener->next;
233 pthread_rwlock_unlock(&listeners_rwlock);
237 * Jobs callback for broadcasting string asynchronously
239 static void broadcast_job(int signum, void *closure)
241 struct job_broadcast *jb = closure;
244 broadcast(jb->event, jb->object);
245 destroy_job_broadcast(jb);
249 * Broadcasts the string 'event' with its 'object'
251 static int unhooked_broadcast(const char *event, struct json_object *object)
253 struct job_broadcast *jb;
256 jb = make_job_broadcast(event, object);
258 ERROR("Cant't create broadcast string job item for %s(%s)",
259 event, json_object_to_json_string(object));
260 json_object_put(object);
264 rc = jobs_queue(BROADCAST_JOB_GROUP, 0, broadcast_job, jb);
266 ERROR("cant't queue broadcast string job item for %s(%s)",
267 event, json_object_to_json_string(object));
268 destroy_job_broadcast(jb);
274 * Broadcasts the event 'evtid' with its 'object'
275 * 'object' is released (like json_object_put)
276 * Returns the count of listener that received the event.
278 int afb_evt_evtid_broadcast(struct afb_evtid *evtid, struct json_object *object)
280 return unhooked_broadcast(evtid->fullname, object);
284 * Broadcasts the event 'evtid' with its 'object'
285 * 'object' is released (like json_object_put)
286 * Returns the count of listener that received the event.
288 int afb_evt_evtid_hooked_broadcast(struct afb_evtid *evtid, struct json_object *object)
292 json_object_get(object);
294 if (evtid->hookflags & afb_hook_flag_evt_broadcast_before)
295 afb_hook_evt_broadcast_before(evtid->fullname, evtid->id, object);
297 result = afb_evt_evtid_broadcast(evtid, object);
299 if (evtid->hookflags & afb_hook_flag_evt_broadcast_after)
300 afb_hook_evt_broadcast_after(evtid->fullname, evtid->id, object, result);
302 json_object_put(object);
308 * Broadcasts the 'event' with its 'object'
309 * 'object' is released (like json_object_put)
310 * Returns the count of listener having receive the event.
312 int afb_evt_broadcast(const char *event, struct json_object *object)
317 json_object_get(object);
318 afb_hook_evt_broadcast_before(event, 0, object);
321 result = unhooked_broadcast(event, object);
324 afb_hook_evt_broadcast_after(event, 0, object, result);
325 json_object_put(object);
331 * Pushes the event 'evtid' with 'obj' to its listeners
332 * Returns the count of listener that received the event.
334 static void push_evtid(struct afb_evtid *evtid, struct json_object *object)
337 struct afb_evt_watch *watch;
338 struct afb_evt_listener *listener;
341 pthread_rwlock_rdlock(&evtid->rwlock);
342 watch = evtid->watchs;
344 listener = watch->listener;
345 assert(listener->itf->push != NULL);
346 if (watch->activity != 0) {
347 listener->itf->push(listener->closure, evtid->fullname, evtid->id, json_object_get(object));
350 watch = watch->next_by_evtid;
352 evtid->has_client = has_client;
353 pthread_rwlock_unlock(&evtid->rwlock);
357 * Jobs callback for pushing evtid asynchronously
359 static void push_job_evtid(int signum, void *closure)
361 struct job_evtid *je = closure;
364 push_evtid(je->evtid, je->object);
365 destroy_job_evtid(je);
369 * Pushes the event 'evtid' with 'obj' to its listeners
370 * 'obj' is released (like json_object_put)
371 * Returns 1 if at least one listener exists or 0 if no listener exists or
372 * -1 in case of error and the event can't be delivered
374 int afb_evt_evtid_push(struct afb_evtid *evtid, struct json_object *object)
376 struct job_evtid *je;
379 je = make_job_evtid(evtid, object);
381 ERROR("Cant't create push evtid job item for %s(%s)",
382 evtid->fullname, json_object_to_json_string(object));
383 json_object_put(object);
387 rc = jobs_queue(PUSH_JOB_GROUP, 0, push_job_evtid, je);
389 rc = evtid->has_client;
391 ERROR("cant't queue push evtid job item for %s(%s)",
392 evtid->fullname, json_object_to_json_string(object));
393 destroy_job_evtid(je);
400 * Pushes the event 'evtid' with 'obj' to its listeners
401 * 'obj' is released (like json_object_put)
402 * Emits calls to hooks.
403 * Returns the count of listener taht received the event.
405 int afb_evt_evtid_hooked_push(struct afb_evtid *evtid, struct json_object *obj)
410 /* lease the object */
411 json_object_get(obj);
413 /* hook before push */
414 if (evtid->hookflags & afb_hook_flag_evt_push_before)
415 afb_hook_evt_push_before(evtid->fullname, evtid->id, obj);
418 result = afb_evt_evtid_push(evtid, obj);
420 /* hook after push */
421 if (evtid->hookflags & afb_hook_flag_evt_push_after)
422 afb_hook_evt_push_after(evtid->fullname, evtid->id, obj, result);
424 /* release the object */
425 json_object_put(obj);
432 static void remove_watch(struct afb_evt_watch *watch)
434 struct afb_evt_watch **prv;
435 struct afb_evtid *evtid;
436 struct afb_evt_listener *listener;
438 /* notify listener if needed */
439 evtid = watch->evtid;
440 listener = watch->listener;
441 if (watch->activity != 0 && listener->itf->remove != NULL)
442 listener->itf->remove(listener->closure, evtid->fullname, evtid->id);
444 /* unlink the watch for its event */
445 prv = &evtid->watchs;
447 prv = &(*prv)->next_by_evtid;
448 *prv = watch->next_by_evtid;
450 /* unlink the watch for its listener */
451 prv = &listener->watchs;
453 prv = &(*prv)->next_by_listener;
454 *prv = watch->next_by_listener;
461 * Creates an event of name 'fullname' and returns it or NULL on error.
463 struct afb_evtid *afb_evt_evtid_create(const char *fullname)
466 struct afb_evtid *evtid, *oevt;
468 /* allocates the event */
469 len = strlen(fullname);
470 evtid = malloc(len + 1 + sizeof * evtid);
474 /* allocates the id */
475 pthread_rwlock_wrlock(&events_rwlock);
477 if (++event_id_counter < 0) {
478 event_id_wrapped = 1;
479 event_id_counter = 1024; /* heuristic: small numbers are not destroyed */
481 if (!event_id_wrapped)
484 while(oevt != NULL && oevt->id != event_id_counter)
486 } while (oevt != NULL);
488 /* initialize the event */
489 memcpy(evtid->fullname, fullname, len + 1);
490 evtid->next = evtids;
492 evtid->watchs = NULL;
493 evtid->id = event_id_counter;
494 evtid->has_client = 0;
495 pthread_rwlock_init(&evtid->rwlock, NULL);
497 evtid->hookflags = afb_hook_flags_evt(evtid->fullname);
498 evtid->eventid.itf = evtid->hookflags ? &afb_evt_hooked_eventid_itf : &afb_evt_event_x2_itf;
499 if (evtid->hookflags & afb_hook_flag_evt_create)
500 afb_hook_evt_create(evtid->fullname, evtid->id);
501 pthread_rwlock_unlock(&events_rwlock);
503 /* returns the event */
510 * Creates an event of name 'prefix'/'name' and returns it or NULL on error.
512 struct afb_evtid *afb_evt_evtid_create2(const char *prefix, const char *name)
514 size_t prelen, postlen;
517 /* makes the event fullname */
518 prelen = strlen(prefix);
519 postlen = strlen(name);
520 fullname = alloca(prelen + postlen + 2);
521 memcpy(fullname, prefix, prelen);
522 fullname[prelen] = '/';
523 memcpy(fullname + prelen + 1, name, postlen + 1);
525 /* create the event */
526 return afb_evt_evtid_create(fullname);
530 * increment the reference count of the event 'evtid'
532 struct afb_evtid *afb_evt_evtid_addref(struct afb_evtid *evtid)
534 __atomic_add_fetch(&evtid->refcount, 1, __ATOMIC_RELAXED);
539 * increment the reference count of the event 'evtid'
541 struct afb_evtid *afb_evt_evtid_hooked_addref(struct afb_evtid *evtid)
543 if (evtid->hookflags & afb_hook_flag_evt_addref)
544 afb_hook_evt_addref(evtid->fullname, evtid->id);
545 return afb_evt_evtid_addref(evtid);
549 * decrement the reference count of the event 'evtid'
550 * and destroy it when the count reachs zero
552 void afb_evt_evtid_unref(struct afb_evtid *evtid)
555 struct afb_evtid **prv;
556 struct afb_evt_listener *listener;
558 if (!__atomic_sub_fetch(&evtid->refcount, 1, __ATOMIC_RELAXED)) {
559 /* unlinks the event if valid! */
560 pthread_rwlock_wrlock(&events_rwlock);
563 while (*prv && !(found = (*prv == evtid)))
567 pthread_rwlock_unlock(&events_rwlock);
569 /* destroys the event */
571 ERROR("event not found");
573 /* removes all watchers */
574 while(evtid->watchs != NULL) {
575 listener = evtid->watchs->listener;
576 pthread_rwlock_wrlock(&listener->rwlock);
577 pthread_rwlock_wrlock(&evtid->rwlock);
578 remove_watch(evtid->watchs);
579 pthread_rwlock_unlock(&evtid->rwlock);
580 pthread_rwlock_unlock(&listener->rwlock);
584 pthread_rwlock_destroy(&evtid->rwlock);
591 * decrement the reference count of the event 'evtid'
592 * and destroy it when the count reachs zero
594 void afb_evt_evtid_hooked_unref(struct afb_evtid *evtid)
596 if (evtid->hookflags & afb_hook_flag_evt_unref)
597 afb_hook_evt_unref(evtid->fullname, evtid->id);
598 afb_evt_evtid_unref(evtid);
602 * Returns the true name of the 'event'
604 const char *afb_evt_evtid_fullname(struct afb_evtid *evtid)
606 return evtid->fullname;
610 * Returns the name of the 'event'
612 const char *afb_evt_evtid_name(struct afb_evtid *evtid)
614 const char *name = strchr(evtid->fullname, '/');
615 return name ? name + 1 : evtid->fullname;
619 * Returns the name associated to the event 'evtid'.
621 const char *afb_evt_evtid_hooked_name(struct afb_evtid *evtid)
623 const char *result = afb_evt_evtid_name(evtid);
624 if (evtid->hookflags & afb_hook_flag_evt_name)
625 afb_hook_evt_name(evtid->fullname, evtid->id, result);
630 * Returns the id of the 'event'
632 int afb_evt_evtid_id(struct afb_evtid *evtid)
638 * Returns an instance of the listener defined by the 'send' callback
640 * Returns NULL in case of memory depletion.
642 struct afb_evt_listener *afb_evt_listener_create(const struct afb_evt_itf *itf, void *closure)
644 struct afb_evt_listener *listener;
646 /* search if an instance already exists */
647 pthread_rwlock_wrlock(&listeners_rwlock);
648 listener = listeners;
649 while (listener != NULL) {
650 if (listener->itf == itf && listener->closure == closure) {
651 listener = afb_evt_listener_addref(listener);
654 listener = listener->next;
658 listener = calloc(1, sizeof *listener);
659 if (listener != NULL) {
662 listener->closure = closure;
663 listener->watchs = NULL;
664 listener->refcount = 1;
665 pthread_rwlock_init(&listener->rwlock, NULL);
666 listener->next = listeners;
667 listeners = listener;
670 pthread_rwlock_unlock(&listeners_rwlock);
675 * Increases the reference count of 'listener' and returns it
677 struct afb_evt_listener *afb_evt_listener_addref(struct afb_evt_listener *listener)
679 __atomic_add_fetch(&listener->refcount, 1, __ATOMIC_RELAXED);
684 * Decreases the reference count of the 'listener' and destroys it
687 void afb_evt_listener_unref(struct afb_evt_listener *listener)
689 struct afb_evt_listener **prv;
690 struct afb_evtid *evtid;
692 if (listener && !__atomic_sub_fetch(&listener->refcount, 1, __ATOMIC_RELAXED)) {
694 /* unlink the listener */
695 pthread_rwlock_wrlock(&listeners_rwlock);
697 while (*prv != listener)
699 *prv = listener->next;
700 pthread_rwlock_unlock(&listeners_rwlock);
702 /* remove the watchers */
703 pthread_rwlock_wrlock(&listener->rwlock);
704 while (listener->watchs != NULL) {
705 evtid = listener->watchs->evtid;
706 pthread_rwlock_wrlock(&evtid->rwlock);
707 remove_watch(listener->watchs);
708 pthread_rwlock_unlock(&evtid->rwlock);
710 pthread_rwlock_unlock(&listener->rwlock);
712 /* free the listener */
713 pthread_rwlock_destroy(&listener->rwlock);
719 * Makes the 'listener' watching 'evtid'
720 * Returns 0 in case of success or else -1.
722 int afb_evt_watch_add_evtid(struct afb_evt_listener *listener, struct afb_evtid *evtid)
724 struct afb_evt_watch *watch;
726 /* check parameter */
727 if (listener->itf->push == NULL) {
732 /* search the existing watch for the listener */
733 pthread_rwlock_wrlock(&listener->rwlock);
734 watch = listener->watchs;
735 while(watch != NULL) {
736 if (watch->evtid == evtid)
738 watch = watch->next_by_listener;
741 /* not found, allocate a new */
742 watch = malloc(sizeof *watch);
744 pthread_rwlock_unlock(&listener->rwlock);
749 /* initialise and link */
750 watch->evtid = evtid;
752 watch->listener = listener;
753 watch->next_by_listener = listener->watchs;
754 listener->watchs = watch;
755 pthread_rwlock_wrlock(&evtid->rwlock);
756 watch->next_by_evtid = evtid->watchs;
757 evtid->watchs = watch;
758 pthread_rwlock_unlock(&evtid->rwlock);
761 if (watch->activity == 0 && listener->itf->add != NULL)
762 listener->itf->add(listener->closure, evtid->fullname, evtid->id);
764 evtid->has_client = 1;
765 pthread_rwlock_unlock(&listener->rwlock);
771 * Avoids the 'listener' to watch 'evtid'
772 * Returns 0 in case of success or else -1.
774 int afb_evt_watch_sub_evtid(struct afb_evt_listener *listener, struct afb_evtid *evtid)
776 struct afb_evt_watch *watch;
778 /* search the existing watch */
779 pthread_rwlock_wrlock(&listener->rwlock);
780 watch = listener->watchs;
781 while(watch != NULL) {
782 if (watch->evtid == evtid) {
783 if (watch->activity != 0) {
785 if (watch->activity == 0 && listener->itf->remove != NULL)
786 listener->itf->remove(listener->closure, evtid->fullname, evtid->id);
788 pthread_rwlock_unlock(&listener->rwlock);
791 watch = watch->next_by_listener;
793 pthread_rwlock_unlock(&listener->rwlock);
799 * update the hooks for events
801 void afb_evt_update_hooks()
803 struct afb_evtid *evtid;
805 pthread_rwlock_rdlock(&events_rwlock);
806 for (evtid = evtids ; evtid ; evtid = evtid->next) {
807 evtid->hookflags = afb_hook_flags_evt(evtid->fullname);
808 evtid->eventid.itf = evtid->hookflags ? &afb_evt_hooked_eventid_itf : &afb_evt_event_x2_itf;
810 pthread_rwlock_unlock(&events_rwlock);
813 inline struct afb_evtid *afb_evt_event_x2_to_evtid(struct afb_event_x2 *eventid)
815 return (struct afb_evtid*)eventid;
818 inline struct afb_event_x2 *afb_evt_event_x2_from_evtid(struct afb_evtid *evtid)
820 return &evtid->eventid;
824 * Creates an event of 'fullname' and returns it.
825 * Returns an event with closure==NULL in case of error.
827 struct afb_event_x2 *afb_evt_event_x2_create(const char *fullname)
829 return afb_evt_event_x2_from_evtid(afb_evt_evtid_create(fullname));
833 * Creates an event of name 'prefix'/'name' and returns it.
834 * Returns an event with closure==NULL in case of error.
836 struct afb_event_x2 *afb_evt_event_x2_create2(const char *prefix, const char *name)
838 return afb_evt_event_x2_from_evtid(afb_evt_evtid_create2(prefix, name));
842 * Returns the fullname of the 'eventid'
844 const char *afb_evt_event_x2_fullname(struct afb_event_x2 *eventid)
846 struct afb_evtid *evtid = afb_evt_event_x2_to_evtid(eventid);
847 return evtid ? evtid->fullname : NULL;
851 * Returns the id of the 'eventid'
853 int afb_evt_event_x2_id(struct afb_event_x2 *eventid)
855 struct afb_evtid *evtid = afb_evt_event_x2_to_evtid(eventid);
856 return evtid ? evtid->id : 0;
860 * Makes the 'listener' watching 'eventid'
861 * Returns 0 in case of success or else -1.
863 int afb_evt_event_x2_add_watch(struct afb_evt_listener *listener, struct afb_event_x2 *eventid)
865 struct afb_evtid *evtid = afb_evt_event_x2_to_evtid(eventid);
867 /* check parameter */
873 /* search the existing watch for the listener */
874 return afb_evt_watch_add_evtid(listener, evtid);
878 * Avoids the 'listener' to watch 'eventid'
879 * Returns 0 in case of success or else -1.
881 int afb_evt_event_x2_remove_watch(struct afb_evt_listener *listener, struct afb_event_x2 *eventid)
883 struct afb_evtid *evtid = afb_evt_event_x2_to_evtid(eventid);
885 /* check parameter */
891 /* search the existing watch */
892 return afb_evt_watch_sub_evtid(listener, evtid);
895 int afb_evt_event_x2_push(struct afb_event_x2 *eventid, struct json_object *object)
897 struct afb_evtid *evtid = afb_evt_event_x2_to_evtid(eventid);
899 return afb_evt_evtid_hooked_push(evtid, object);
900 json_object_put(object);
904 int afb_evt_event_x2_unhooked_push(struct afb_event_x2 *eventid, struct json_object *object)
906 struct afb_evtid *evtid = afb_evt_event_x2_to_evtid(eventid);
908 return afb_evt_evtid_push(evtid, object);
909 json_object_put(object);
913 struct afb_event_x1 afb_evt_event_from_evtid(struct afb_evtid *evtid)
916 ? (struct afb_event_x1){ .itf = &afb_evt_hooked_eventid_itf, .closure = &evtid->eventid }
917 : (struct afb_event_x1){ .itf = NULL, .closure = NULL };
920 void afb_evt_event_x2_unref(struct afb_event_x2 *eventid)
922 struct afb_evtid *evtid = afb_evt_event_x2_to_evtid(eventid);
924 afb_evt_evtid_unref(evtid);
927 struct afb_event_x2 *afb_evt_event_x2_addref(struct afb_event_x2 *eventid)
929 struct afb_evtid *evtid = afb_evt_event_x2_to_evtid(eventid);
931 afb_evt_evtid_addref(evtid);