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