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                 prv = &events;
 238                 while (*prv && !(found = (*prv == evt)))
 239                         prv = &(*prv)->next;
 240                 if (found)
 241                         *prv = evt->next;
 242                 pthread_mutex_unlock(&events_mutex);
 243
 244                 /* destroys the event */
 245                 if (found) {
 246                         /* removes all watchers */
 247                         while(evt->watchs != NULL) {
 248                                 listener = evt->watchs->listener;
 249                                 pthread_mutex_lock(&listener->mutex);
 250                                 pthread_mutex_lock(&evt->mutex);
 251                                 remove_watch(evt->watchs);
 252                                 pthread_mutex_unlock(&evt->mutex);
 253                                 pthread_mutex_unlock(&listener->mutex);
 254                         }
 255
 256                         /* free */
 257                         pthread_mutex_destroy(&evt->mutex);
 258                         free(evt);
 259                 }
 260         }
 261 }
 262
 263 /*
 264  * Creates an event of 'name' and returns it.
 265  * Returns an event with closure==NULL in case of error.
 266  */
 267 struct afb_event afb_evt_create_event(const char *name)
 268 {
 269         size_t len;
 270         struct afb_evt_event *evt;
 271
 272         /* allocates the event */
 273         len = strlen(name);
 274         evt = malloc(len + sizeof * evt);
 275         if (evt == NULL)
 276                 goto error;
 277
 278         /* initialize the event */
 279         evt->watchs = NULL;
 280         memcpy(evt->name, name, len + 1);
 281
 282         /* allocates the id */
 283         pthread_mutex_lock(&events_mutex);
 284         do {
 285                 if (++event_id_counter < 0) {
 286                         event_id_wrapped = 1;
 287                         event_id_counter = 1024; /* heuristic: small numbers are not destroyed */
 288                 }
 289                 if (!event_id_wrapped)
 290                         break;
 291                 evt = events;
 292                 while(evt != NULL && evt->id != event_id_counter)
 293                         evt = evt->next;
 294         } while (evt != NULL);
 295
 296         /* initialize the event */
 297         memcpy(evt->name, name, len + 1);
 298         evt->next = events;
 299         evt->watchs = NULL;
 300         evt->id = event_id_counter;
 301         pthread_mutex_init(&evt->mutex, NULL);
 302         events = evt;
 303         pthread_mutex_unlock(&events_mutex);
 304
 305         /* returns the event */
 306         return (struct afb_event){ .itf = &afb_evt_event_itf, .closure = evt };
 307 error:
 308         return (struct afb_event){ .itf = NULL, .closure = NULL };
 309 }
 310
 311 /*
 312  * Returns the name of the 'event'
 313  */
 314 const char *afb_evt_event_name(struct afb_event event)
 315 {
 316         return (event.itf != &afb_evt_event_itf) ? NULL : ((struct afb_evt_event *)event.closure)->name;
 317 }
 318
 319 /*
 320  * Returns the id of the 'event'
 321  */
 322 int afb_evt_event_id(struct afb_event event)
 323 {
 324         return (event.itf != &afb_evt_event_itf) ? 0 : ((struct afb_evt_event *)event.closure)->id;
 325 }
 326
 327 /*
 328  * Returns an instance of the listener defined by the 'send' callback
 329  * and the 'closure'.
 330  * Returns NULL in case of memory depletion.
 331  */
 332 struct afb_evt_listener *afb_evt_listener_create(const struct afb_evt_itf *itf, void *closure)
 333 {
 334         struct afb_evt_listener *listener;
 335
 336         /* search if an instance already exists */
 337         pthread_mutex_lock(&listeners_mutex);
 338         listener = listeners;
 339         while (listener != NULL) {
 340                 if (listener->itf == itf && listener->closure == closure) {
 341                         listener = afb_evt_listener_addref(listener);
 342                         goto found;
 343                 }
 344                 listener = listener->next;
 345         }
 346
 347         /* allocates */
 348         listener = calloc(1, sizeof *listener);
 349         if (listener != NULL) {
 350                 /* init */
 351                 listener->itf = itf;
 352                 listener->closure = closure;
 353                 listener->watchs = NULL;
 354                 listener->refcount = 1;
 355                 pthread_mutex_init(&listener->mutex, NULL);
 356                 listener->next = listeners;
 357                 listeners = listener;
 358         }
 359  found:
 360         pthread_mutex_unlock(&listeners_mutex);
 361         return listener;
 362 }
 363
 364 /*
 365  * Increases the reference count of 'listener' and returns it
 366  */
 367 struct afb_evt_listener *afb_evt_listener_addref(struct afb_evt_listener *listener)
 368 {
 369         __atomic_add_fetch(&listener->refcount, 1, __ATOMIC_RELAXED);
 370         return listener;
 371 }
 372
 373 /*
 374  * Decreases the reference count of the 'listener' and destroys it
 375  * when no more used.
 376  */
 377 void afb_evt_listener_unref(struct afb_evt_listener *listener)
 378 {
 379         struct afb_evt_listener **prv;
 380         struct afb_evt_event *evt;
 381
 382         if (!__atomic_sub_fetch(&listener->refcount, 1, __ATOMIC_RELAXED)) {
 383
 384                 /* unlink the listener */
 385                 pthread_mutex_lock(&listeners_mutex);
 386                 prv = &listeners;
 387                 while (*prv != listener)
 388                         prv = &(*prv)->next;
 389                 *prv = listener->next;
 390                 pthread_mutex_unlock(&listeners_mutex);
 391
 392                 /* remove the watchers */
 393                 pthread_mutex_lock(&listener->mutex);
 394                 while (listener->watchs != NULL) {
 395                         evt = listener->watchs->event;
 396                         pthread_mutex_lock(&evt->mutex);
 397                         remove_watch(listener->watchs);
 398                         pthread_mutex_unlock(&evt->mutex);
 399                 }
 400                 pthread_mutex_unlock(&listener->mutex);
 401
 402                 /* free the listener */
 403                 pthread_mutex_destroy(&listener->mutex);
 404                 free(listener);
 405         }
 406 }
 407
 408 /*
 409  * Makes the 'listener' watching 'event'
 410  * Returns 0 in case of success or else -1.
 411  */
 412 int afb_evt_add_watch(struct afb_evt_listener *listener, struct afb_event event)
 413 {
 414         struct afb_evt_watch *watch;
 415         struct afb_evt_event *evt;
 416
 417         /* check parameter */
 418         if (event.itf != &afb_evt_event_itf || listener->itf->push == NULL) {
 419                 errno = EINVAL;
 420                 return -1;
 421         }
 422
 423         /* search the existing watch for the listener */
 424         evt = event.closure;
 425         pthread_mutex_lock(&listener->mutex);
 426         watch = listener->watchs;
 427         while(watch != NULL) {
 428                 if (watch->event == evt)
 429                         goto found;
 430                 watch = watch->next_by_listener;
 431         }
 432
 433         /* not found, allocate a new */
 434         watch = malloc(sizeof *watch);
 435         if (watch == NULL) {
 436                 pthread_mutex_unlock(&listener->mutex);
 437                 errno = ENOMEM;
 438                 return -1;
 439         }
 440
 441         /* initialise and link */
 442         watch->event = evt;
 443         watch->activity = 0;
 444         watch->listener = listener;
 445         watch->next_by_listener = listener->watchs;
 446         listener->watchs = watch;
 447         pthread_mutex_lock(&evt->mutex);
 448         watch->next_by_event = evt->watchs;
 449         evt->watchs = watch;
 450         pthread_mutex_unlock(&evt->mutex);
 451
 452 found:
 453         if (watch->activity == 0 && listener->itf->add != NULL)
 454                 listener->itf->add(listener->closure, evt->name, evt->id);
 455         watch->activity++;
 456         pthread_mutex_unlock(&listener->mutex);
 457
 458         return 0;
 459 }
 460
 461 /*
 462  * Avoids the 'listener' to watch 'event'
 463  * Returns 0 in case of success or else -1.
 464  */
 465 int afb_evt_remove_watch(struct afb_evt_listener *listener, struct afb_event event)
 466 {
 467         struct afb_evt_watch *watch;
 468         struct afb_evt_event *evt;
 469
 470         /* check parameter */
 471         if (event.itf != &afb_evt_event_itf) {
 472                 errno = EINVAL;
 473                 return -1;
 474         }
 475
 476         /* search the existing watch */
 477         evt = event.closure;
 478         pthread_mutex_lock(&listener->mutex);
 479         watch = listener->watchs;
 480         while(watch != NULL) {
 481                 if (watch->event == evt) {
 482                         /* found: remove it */
 483                         if (watch->activity != 0) {
 484                                 watch->activity--;
 485                                 if (watch->activity == 0 && listener->itf->remove != NULL)
 486                                         listener->itf->remove(listener->closure, evt->name, evt->id);
 487                         }
 488                         pthread_mutex_unlock(&listener->mutex);
 489                         return 0;
 490                 }
 491                 watch = watch->next_by_listener;
 492         }
 493         pthread_mutex_unlock(&listener->mutex);
 494         errno = ENOENT;
 495         return -1;
 496 }
 497