src/afb-evt.c

   1 /*
   2  * Copyright (C) 2015, 2016, 2017 "IoT.bzh"
   3  * Author "Fulup Ar Foll"
   4  * Author José Bollo <jose.bollo@iot.bzh>
   5  *
   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
   9  *
  10  *   http://www.apache.org/licenses/LICENSE-2.0
  11  *
  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.
  17  */
  18
  19 #define _GNU_SOURCE
  20
  21 #include <stdlib.h>
  22 #include <string.h>
  23 #include <assert.h>
  24 #include <errno.h>
  25 #include <pthread.h>
  26
  27 #include <json-c/json.h>
  28 #include <afb/afb-event-itf.h>
  29
  30 #include "afb-evt.h"
  31
  32 struct afb_evt_watch;
  33
  34 /*
  35  * Structure for event listeners
  36  */
  37 struct afb_evt_listener {
  38
  39         /* chaining listeners */
  40         struct afb_evt_listener *next;
  41
  42         /* interface for callbacks */
  43         const struct afb_evt_itf *itf;
  44
  45         /* closure for the callback */
  46         void *closure;
  47
  48         /* head of the list of events listened */
  49         struct afb_evt_watch *watchs;
  50
  51         /* mutex of the listener */
  52         pthread_mutex_t mutex;
  53
  54         /* count of reference to the listener */
  55         int refcount;
  56 };
  57
  58 /*
  59  * Structure for describing events
  60  */
  61 struct afb_evt_event {
  62
  63         /* next event */
  64         struct afb_evt_event *next;
  65
  66         /* head of the list of listeners watching the event */
  67         struct afb_evt_watch *watchs;
  68
  69         /* id of the event */
  70         int id;
  71
  72         /* mutex of the event */
  73         pthread_mutex_t mutex;
  74
  75         /* name of the event */
  76         char name[1];
  77 };
  78
  79 /*
  80  * Structure for associating events and listeners
  81  */
  82 struct afb_evt_watch {
  83
  84         /* the event */
  85         struct afb_evt_event *event;
  86
  87         /* link to the next listener for the same event */
  88         struct afb_evt_watch *next_by_event;
  89
  90         /* the listener */
  91         struct afb_evt_listener *listener;
  92
  93         /* link to the next event for the same listener */
  94         struct afb_evt_watch *next_by_listener;
  95
  96         /* activity */
  97         unsigned activity;
  98 };
  99
 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);
 105
 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
 112 };
 113
 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;
 117
 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;
 123
 124 /*
 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.
 128  */
 129 static int evt_broadcast(struct afb_evt_event *evt, struct json_object *object)
 130 {
 131         return afb_evt_broadcast(evt->name, object);
 132 }
 133
 134 /*
 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.
 138  */
 139 int afb_evt_broadcast(const char *event, struct json_object *object)
 140 {
 141         int result;
 142         struct afb_evt_listener *listener;
 143
 144         result = 0;
 145         pthread_mutex_lock(&listeners_mutex);
 146         listener = listeners;
 147         while(listener) {
 148                 if (listener->itf->broadcast != NULL) {
 149                         listener->itf->broadcast(listener->closure, event, 0, json_object_get(object));
 150                         result++;
 151                 }
 152                 listener = listener->next;
 153         }
 154         pthread_mutex_unlock(&listeners_mutex);
 155         json_object_put(object);
 156         return result;
 157 }
 158
 159 /*
 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.
 163  */
 164 static int evt_push(struct afb_evt_event *evt, struct json_object *obj)
 165 {
 166         int result;
 167         struct afb_evt_watch *watch;
 168         struct afb_evt_listener *listener;
 169
 170         result = 0;
 171         pthread_mutex_lock(&evt->mutex);
 172         watch = evt->watchs;
 173         while(watch) {
 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;
 179                 result++;
 180         }
 181         pthread_mutex_unlock(&evt->mutex);
 182         json_object_put(obj);
 183         return result;
 184 }
 185
 186 /*
 187  * Returns the name associated to the event 'evt'.
 188  */
 189 static const char *evt_name(struct afb_evt_event *evt)
 190 {
 191         return evt->name;
 192 }
 193
 194 /*
 195  * remove the 'watch'
 196  */
 197 static void remove_watch(struct afb_evt_watch *watch)
 198 {
 199         struct afb_evt_watch **prv;
 200         struct afb_evt_event *evt;
 201         struct afb_evt_listener *listener;
 202
 203         /* notify listener if needed */
 204         evt = watch->event;
 205         listener = watch->listener;
 206         if (watch->activity != 0 && listener->itf->remove != NULL)
 207                 listener->itf->remove(listener->closure, evt->name, evt->id);
 208
 209         /* unlink the watch for its event */
 210         prv = &evt->watchs;
 211         while(*prv != watch)
 212                 prv = &(*prv)->next_by_event;
 213         *prv = watch->next_by_event;
 214
 215         /* unlink the watch for its listener */
 216         prv = &listener->watchs;
 217         while(*prv != watch)
 218                 prv = &(*prv)->next_by_listener;
 219         *prv = watch->next_by_listener;
 220
 221         /* recycle memory */
 222         free(watch);
 223 }
 224
 225 /*
 226  * Destroys the event 'evt'
 227  */
 228 static void evt_destroy(struct afb_evt_event *evt)
 229 {
 230         int found;
 231         struct afb_evt_event **prv;
 232         struct afb_evt_listener *listener;
 233
 234         if (evt != NULL) {
 235                 /* unlinks the event if valid! */
 236                 pthread_mutex_lock(&events_mutex);
 237                 found = 0;
 238                 prv = &events;
 239                 while (*prv && !(found = (*prv == evt)))
 240                         prv = &(*prv)->next;
 241                 if (found)
 242                         *prv = evt->next;
 243                 pthread_mutex_unlock(&events_mutex);
 244
 245                 /* destroys the event */
 246                 if (found) {
 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);
 255                         }
 256
 257                         /* free */
 258                         pthread_mutex_destroy(&evt->mutex);
 259                         free(evt);
 260                 }
 261         }
 262 }
 263
 264 /*
 265  * Creates an event of 'name' and returns it.
 266  * Returns an event with closure==NULL in case of error.
 267  */
 268 struct afb_event afb_evt_create_event(const char *name)
 269 {
 270         size_t len;
 271         struct afb_evt_event *evt;
 272
 273         /* allocates the event */
 274         len = strlen(name);
 275         evt = malloc(len + sizeof * evt);
 276         if (evt == NULL)
 277                 goto error;
 278
 279         /* initialize the event */
 280         evt->watchs = NULL;
 281         memcpy(evt->name, name, len + 1);
 282
 283         /* allocates the id */
 284         pthread_mutex_lock(&events_mutex);
 285         do {
 286                 if (++event_id_counter < 0) {
 287                         event_id_wrapped = 1;
 288                         event_id_counter = 1024; /* heuristic: small numbers are not destroyed */
 289                 }
 290                 if (!event_id_wrapped)
 291                         break;
 292                 evt = events;
 293                 while(evt != NULL && evt->id != event_id_counter)
 294                         evt = evt->next;
 295         } while (evt != NULL);
 296
 297         /* initialize the event */
 298         memcpy(evt->name, name, len + 1);
 299         evt->next = events;
 300         evt->watchs = NULL;
 301         evt->id = event_id_counter;
 302         pthread_mutex_init(&evt->mutex, NULL);
 303         events = evt;
 304         pthread_mutex_unlock(&events_mutex);
 305
 306         /* returns the event */
 307         return (struct afb_event){ .itf = &afb_evt_event_itf, .closure = evt };
 308 error:
 309         return (struct afb_event){ .itf = NULL, .closure = NULL };
 310 }
 311
 312 /*
 313  * Returns the name of the 'event'
 314  */
 315 const char *afb_evt_event_name(struct afb_event event)
 316 {
 317         return (event.itf != &afb_evt_event_itf) ? NULL : ((struct afb_evt_event *)event.closure)->name;
 318 }
 319
 320 /*
 321  * Returns the id of the 'event'
 322  */
 323 int afb_evt_event_id(struct afb_event event)
 324 {
 325         return (event.itf != &afb_evt_event_itf) ? 0 : ((struct afb_evt_event *)event.closure)->id;
 326 }
 327
 328 /*
 329  * Returns an instance of the listener defined by the 'send' callback
 330  * and the 'closure'.
 331  * Returns NULL in case of memory depletion.
 332  */
 333 struct afb_evt_listener *afb_evt_listener_create(const struct afb_evt_itf *itf, void *closure)
 334 {
 335         struct afb_evt_listener *listener;
 336
 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);
 343                         goto found;
 344                 }
 345                 listener = listener->next;
 346         }
 347
 348         /* allocates */
 349         listener = calloc(1, sizeof *listener);
 350         if (listener != NULL) {
 351                 /* init */
 352                 listener->itf = itf;
 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;
 359         }
 360  found:
 361         pthread_mutex_unlock(&listeners_mutex);
 362         return listener;
 363 }
 364
 365 /*
 366  * Increases the reference count of 'listener' and returns it
 367  */
 368 struct afb_evt_listener *afb_evt_listener_addref(struct afb_evt_listener *listener)
 369 {
 370         __atomic_add_fetch(&listener->refcount, 1, __ATOMIC_RELAXED);
 371         return listener;
 372 }
 373
 374 /*
 375  * Decreases the reference count of the 'listener' and destroys it
 376  * when no more used.
 377  */
 378 void afb_evt_listener_unref(struct afb_evt_listener *listener)
 379 {
 380         struct afb_evt_listener **prv;
 381         struct afb_evt_event *evt;
 382
 383         if (!__atomic_sub_fetch(&listener->refcount, 1, __ATOMIC_RELAXED)) {
 384
 385                 /* unlink the listener */
 386                 pthread_mutex_lock(&listeners_mutex);
 387                 prv = &listeners;
 388                 while (*prv != listener)
 389                         prv = &(*prv)->next;
 390                 *prv = listener->next;
 391                 pthread_mutex_unlock(&listeners_mutex);
 392
 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);
 400                 }
 401                 pthread_mutex_unlock(&listener->mutex);
 402
 403                 /* free the listener */
 404                 pthread_mutex_destroy(&listener->mutex);
 405                 free(listener);
 406         }
 407 }
 408
 409 /*
 410  * Makes the 'listener' watching 'event'
 411  * Returns 0 in case of success or else -1.
 412  */
 413 int afb_evt_add_watch(struct afb_evt_listener *listener, struct afb_event event)
 414 {
 415         struct afb_evt_watch *watch;
 416         struct afb_evt_event *evt;
 417
 418         /* check parameter */
 419         if (event.itf != &afb_evt_event_itf || listener->itf->push == NULL) {
 420                 errno = EINVAL;
 421                 return -1;
 422         }
 423
 424         /* search the existing watch for the listener */
 425         evt = event.closure;
 426         pthread_mutex_lock(&listener->mutex);
 427         watch = listener->watchs;
 428         while(watch != NULL) {
 429                 if (watch->event == evt)
 430                         goto found;
 431                 watch = watch->next_by_listener;
 432         }
 433
 434         /* not found, allocate a new */
 435         watch = malloc(sizeof *watch);
 436         if (watch == NULL) {
 437                 pthread_mutex_unlock(&listener->mutex);
 438                 errno = ENOMEM;
 439                 return -1;
 440         }
 441
 442         /* initialise and link */
 443         watch->event = evt;
 444         watch->activity = 0;
 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;
 450         evt->watchs = watch;
 451         pthread_mutex_unlock(&evt->mutex);
 452
 453 found:
 454         if (watch->activity == 0 && listener->itf->add != NULL)
 455                 listener->itf->add(listener->closure, evt->name, evt->id);
 456         watch->activity++;
 457         pthread_mutex_unlock(&listener->mutex);
 458
 459         return 0;
 460 }
 461
 462 /*
 463  * Avoids the 'listener' to watch 'event'
 464  * Returns 0 in case of success or else -1.
 465  */
 466 int afb_evt_remove_watch(struct afb_evt_listener *listener, struct afb_event event)
 467 {
 468         struct afb_evt_watch *watch;
 469         struct afb_evt_event *evt;
 470
 471         /* check parameter */
 472         if (event.itf != &afb_evt_event_itf) {
 473                 errno = EINVAL;
 474                 return -1;
 475         }
 476
 477         /* search the existing watch */
 478         evt = event.closure;
 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) {
 485                                 watch->activity--;
 486                                 if (watch->activity == 0 && listener->itf->remove != NULL)
 487                                         listener->itf->remove(listener->closure, evt->name, evt->id);
 488                         }
 489                         pthread_mutex_unlock(&listener->mutex);
 490                         return 0;
 491                 }
 492                 watch = watch->next_by_listener;
 493         }
 494         pthread_mutex_unlock(&listener->mutex);
 495         errno = ENOENT;
 496         return -1;
 497 }
 498