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 "afb-hook.h"
  32 #include "verbose.h"
  33
  34 struct afb_evt_watch;
  35
  36 /*
  37  * Structure for event listeners
  38  */
  39 struct afb_evt_listener {
  40
  41         /* chaining listeners */
  42         struct afb_evt_listener *next;
  43
  44         /* interface for callbacks */
  45         const struct afb_evt_itf *itf;
  46
  47         /* closure for the callback */
  48         void *closure;
  49
  50         /* head of the list of events listened */
  51         struct afb_evt_watch *watchs;
  52
  53         /* mutex of the listener */
  54         pthread_mutex_t mutex;
  55
  56         /* count of reference to the listener */
  57         int refcount;
  58 };
  59
  60 /*
  61  * Structure for describing events
  62  */
  63 struct afb_evt_event {
  64
  65         /* next event */
  66         struct afb_evt_event *next;
  67
  68         /* head of the list of listeners watching the event */
  69         struct afb_evt_watch *watchs;
  70
  71         /* id of the event */
  72         int id;
  73
  74         /* hooking */
  75         int hookflags;
  76
  77         /* mutex of the event */
  78         pthread_mutex_t mutex;
  79
  80         /* name of the event */
  81         char name[1];
  82 };
  83
  84 /*
  85  * Structure for associating events and listeners
  86  */
  87 struct afb_evt_watch {
  88
  89         /* the event */
  90         struct afb_evt_event *event;
  91
  92         /* link to the next listener for the same event */
  93         struct afb_evt_watch *next_by_event;
  94
  95         /* the listener */
  96         struct afb_evt_listener *listener;
  97
  98         /* link to the next event for the same listener */
  99         struct afb_evt_watch *next_by_listener;
 100
 101         /* activity */
 102         unsigned activity;
 103 };
 104
 105 /* declare functions */
 106 static int evt_broadcast(struct afb_evt_event *evt, struct json_object *obj);
 107 static int evt_push(struct afb_evt_event *evt, struct json_object *obj);
 108 static void evt_destroy(struct afb_evt_event *evt);
 109 static const char *evt_name(struct afb_evt_event *evt);
 110
 111 /* the interface for events */
 112 static struct afb_event_itf afb_evt_event_itf = {
 113         .broadcast = (void*)evt_broadcast,
 114         .push = (void*)evt_push,
 115         .drop = (void*)evt_destroy,
 116         .name = (void*)evt_name
 117 };
 118
 119 /* head of the list of listeners */
 120 static pthread_mutex_t listeners_mutex = PTHREAD_MUTEX_INITIALIZER;
 121 static struct afb_evt_listener *listeners = NULL;
 122
 123 /* handling id of events */
 124 static pthread_mutex_t events_mutex = PTHREAD_MUTEX_INITIALIZER;
 125 static struct afb_evt_event *events = NULL;
 126 static int event_id_counter = 0;
 127 static int event_id_wrapped = 0;
 128
 129 /*
 130  * Broadcasts the 'event' of 'id' with its 'obj'
 131  * 'obj' is released (like json_object_put)
 132  * calls hooks if hookflags isn't 0
 133  * Returns the count of listener having receive the event.
 134  */
 135 static int broadcast(const char *event, struct json_object *obj, int id, int hookflags)
 136 {
 137         int result;
 138         struct afb_evt_listener *listener;
 139
 140         if (hookflags & afb_hook_flag_evt_broadcast_before)
 141                 afb_hook_evt_broadcast_before(event, id, obj);
 142         result = 0;
 143         pthread_mutex_lock(&listeners_mutex);
 144         listener = listeners;
 145         while(listener) {
 146                 if (listener->itf->broadcast != NULL) {
 147                         listener->itf->broadcast(listener->closure, event, id, json_object_get(obj));
 148                         result++;
 149                 }
 150                 listener = listener->next;
 151         }
 152         pthread_mutex_unlock(&listeners_mutex);
 153         if (hookflags & afb_hook_flag_evt_broadcast_after)
 154                 afb_hook_evt_broadcast_after(event, id, obj, result);
 155         json_object_put(obj);
 156         return result;
 157 }
 158
 159 /*
 160  * Broadcasts the event 'evt' with its 'object'
 161  * 'object' is released (like json_object_put)
 162  * Returns the count of listener that received the event.
 163  */
 164 static int evt_broadcast(struct afb_evt_event *evt, struct json_object *object)
 165 {
 166         return broadcast(evt->name, object, evt->id, evt->hookflags);
 167 }
 168
 169 /*
 170  * Broadcasts the 'event' with its 'object'
 171  * 'object' is released (like json_object_put)
 172  * Returns the count of listener having receive the event.
 173  */
 174 int afb_evt_broadcast(const char *event, struct json_object *object)
 175 {
 176         return broadcast(event, object, 0, -1);
 177 }
 178
 179 /*
 180  * Pushes the event 'evt' with 'obj' to its listeners
 181  * 'obj' is released (like json_object_put)
 182  * calls hooks if hookflags isn't 0
 183  * Returns the count of listener taht received the event.
 184  */
 185 static int push(struct afb_evt_event *evt, struct json_object *obj, int hookflags)
 186 {
 187         int result;
 188         struct afb_evt_watch *watch;
 189         struct afb_evt_listener *listener;
 190
 191         result = 0;
 192         pthread_mutex_lock(&evt->mutex);
 193         if (hookflags & afb_hook_flag_evt_push_before)
 194                 afb_hook_evt_push_before(evt->name, evt->id, obj);
 195         watch = evt->watchs;
 196         while(watch) {
 197                 listener = watch->listener;
 198                 assert(listener->itf->push != NULL);
 199                 if (watch->activity != 0) {
 200                         listener->itf->push(listener->closure, evt->name, evt->id, json_object_get(obj));
 201                         result++;
 202                 }
 203                 watch = watch->next_by_event;
 204         }
 205         if (hookflags & afb_hook_flag_evt_push_after)
 206                 afb_hook_evt_push_after(evt->name, evt->id, obj, result);
 207         pthread_mutex_unlock(&evt->mutex);
 208         json_object_put(obj);
 209         return result;
 210 }
 211
 212 /*
 213  * Pushes the event 'evt' with 'obj' to its listeners
 214  * 'obj' is released (like json_object_put)
 215  * Returns the count of listener taht received the event.
 216  */
 217 static int evt_push(struct afb_evt_event *evt, struct json_object *obj)
 218 {
 219         return push(evt, obj, evt->hookflags);
 220 }
 221
 222 /*
 223  * Returns the name associated to the event 'evt'.
 224  */
 225 static const char *evt_name(struct afb_evt_event *evt)
 226 {
 227         const char *name = strchr(evt->name, '/');
 228         name = name ? name + 1 : evt->name;
 229         if (evt->hookflags & afb_hook_flag_evt_name)
 230                 afb_hook_evt_name(evt->name, evt->id);
 231         return name;
 232 }
 233
 234 /*
 235  * remove the 'watch'
 236  */
 237 static void remove_watch(struct afb_evt_watch *watch)
 238 {
 239         struct afb_evt_watch **prv;
 240         struct afb_evt_event *evt;
 241         struct afb_evt_listener *listener;
 242
 243         /* notify listener if needed */
 244         evt = watch->event;
 245         listener = watch->listener;
 246         if (watch->activity != 0 && listener->itf->remove != NULL)
 247                 listener->itf->remove(listener->closure, evt->name, evt->id);
 248
 249         /* unlink the watch for its event */
 250         prv = &evt->watchs;
 251         while(*prv != watch)
 252                 prv = &(*prv)->next_by_event;
 253         *prv = watch->next_by_event;
 254
 255         /* unlink the watch for its listener */
 256         prv = &listener->watchs;
 257         while(*prv != watch)
 258                 prv = &(*prv)->next_by_listener;
 259         *prv = watch->next_by_listener;
 260
 261         /* recycle memory */
 262         free(watch);
 263 }
 264
 265 /*
 266  * Destroys the event 'evt'
 267  */
 268 static void evt_destroy(struct afb_evt_event *evt)
 269 {
 270         int found;
 271         struct afb_evt_event **prv;
 272         struct afb_evt_listener *listener;
 273
 274         if (evt != NULL) {
 275                 /* unlinks the event if valid! */
 276                 pthread_mutex_lock(&events_mutex);
 277                 found = 0;
 278                 prv = &events;
 279                 while (*prv && !(found = (*prv == evt)))
 280                         prv = &(*prv)->next;
 281                 if (found)
 282                         *prv = evt->next;
 283                 pthread_mutex_unlock(&events_mutex);
 284
 285                 /* destroys the event */
 286                 if (found) {
 287                         /* removes all watchers */
 288                         while(evt->watchs != NULL) {
 289                                 listener = evt->watchs->listener;
 290                                 pthread_mutex_lock(&listener->mutex);
 291                                 pthread_mutex_lock(&evt->mutex);
 292                                 remove_watch(evt->watchs);
 293                                 pthread_mutex_unlock(&evt->mutex);
 294                                 pthread_mutex_unlock(&listener->mutex);
 295                         }
 296
 297                         /* hook */
 298                         if (evt->hookflags & afb_hook_flag_evt_drop)
 299                                 afb_hook_evt_drop(evt->name, evt->id);
 300
 301                         /* free */
 302                         pthread_mutex_destroy(&evt->mutex);
 303                         free(evt);
 304                 }
 305         }
 306 }
 307
 308 /*
 309  * Creates an event of 'name' and returns it.
 310  * Returns an event with closure==NULL in case of error.
 311  */
 312 struct afb_event afb_evt_create_event(const char *name)
 313 {
 314         size_t len;
 315         struct afb_evt_event *evt;
 316
 317         /* allocates the event */
 318         len = strlen(name);
 319         evt = malloc(len + sizeof * evt);
 320         if (evt == NULL)
 321                 goto error;
 322
 323         /* initialize the event */
 324         evt->watchs = NULL;
 325         memcpy(evt->name, name, len + 1);
 326
 327         /* allocates the id */
 328         pthread_mutex_lock(&events_mutex);
 329         do {
 330                 if (++event_id_counter < 0) {
 331                         event_id_wrapped = 1;
 332                         event_id_counter = 1024; /* heuristic: small numbers are not destroyed */
 333                 }
 334                 if (!event_id_wrapped)
 335                         break;
 336                 evt = events;
 337                 while(evt != NULL && evt->id != event_id_counter)
 338                         evt = evt->next;
 339         } while (evt != NULL);
 340
 341         /* initialize the event */
 342         memcpy(evt->name, name, len + 1);
 343         evt->next = events;
 344         evt->watchs = NULL;
 345         evt->id = event_id_counter;
 346         pthread_mutex_init(&evt->mutex, NULL);
 347         events = evt;
 348         evt->hookflags = afb_hook_flags_evt(evt->name);
 349         if (evt->hookflags & afb_hook_flag_evt_create)
 350                 afb_hook_evt_create(evt->name, evt->id);
 351         pthread_mutex_unlock(&events_mutex);
 352
 353         /* returns the event */
 354         return (struct afb_event){ .itf = &afb_evt_event_itf, .closure = evt };
 355 error:
 356         return (struct afb_event){ .itf = NULL, .closure = NULL };
 357 }
 358
 359 /*
 360  * Returns the name of the 'event'
 361  */
 362 const char *afb_evt_event_name(struct afb_event event)
 363 {
 364         return (event.itf != &afb_evt_event_itf) ? NULL : ((struct afb_evt_event *)event.closure)->name;
 365 }
 366
 367 /*
 368  * Returns the id of the 'event'
 369  */
 370 int afb_evt_event_id(struct afb_event event)
 371 {
 372         return (event.itf != &afb_evt_event_itf) ? 0 : ((struct afb_evt_event *)event.closure)->id;
 373 }
 374
 375 /*
 376  * Returns an instance of the listener defined by the 'send' callback
 377  * and the 'closure'.
 378  * Returns NULL in case of memory depletion.
 379  */
 380 struct afb_evt_listener *afb_evt_listener_create(const struct afb_evt_itf *itf, void *closure)
 381 {
 382         struct afb_evt_listener *listener;
 383
 384         /* search if an instance already exists */
 385         pthread_mutex_lock(&listeners_mutex);
 386         listener = listeners;
 387         while (listener != NULL) {
 388                 if (listener->itf == itf && listener->closure == closure) {
 389                         listener = afb_evt_listener_addref(listener);
 390                         goto found;
 391                 }
 392                 listener = listener->next;
 393         }
 394
 395         /* allocates */
 396         listener = calloc(1, sizeof *listener);
 397         if (listener != NULL) {
 398                 /* init */
 399                 listener->itf = itf;
 400                 listener->closure = closure;
 401                 listener->watchs = NULL;
 402                 listener->refcount = 1;
 403                 pthread_mutex_init(&listener->mutex, NULL);
 404                 listener->next = listeners;
 405                 listeners = listener;
 406         }
 407  found:
 408         pthread_mutex_unlock(&listeners_mutex);
 409         return listener;
 410 }
 411
 412 /*
 413  * Increases the reference count of 'listener' and returns it
 414  */
 415 struct afb_evt_listener *afb_evt_listener_addref(struct afb_evt_listener *listener)
 416 {
 417         __atomic_add_fetch(&listener->refcount, 1, __ATOMIC_RELAXED);
 418         return listener;
 419 }
 420
 421 /*
 422  * Decreases the reference count of the 'listener' and destroys it
 423  * when no more used.
 424  */
 425 void afb_evt_listener_unref(struct afb_evt_listener *listener)
 426 {
 427         struct afb_evt_listener **prv;
 428         struct afb_evt_event *evt;
 429
 430         if (!__atomic_sub_fetch(&listener->refcount, 1, __ATOMIC_RELAXED)) {
 431
 432                 /* unlink the listener */
 433                 pthread_mutex_lock(&listeners_mutex);
 434                 prv = &listeners;
 435                 while (*prv != listener)
 436                         prv = &(*prv)->next;
 437                 *prv = listener->next;
 438                 pthread_mutex_unlock(&listeners_mutex);
 439
 440                 /* remove the watchers */
 441                 pthread_mutex_lock(&listener->mutex);
 442                 while (listener->watchs != NULL) {
 443                         evt = listener->watchs->event;
 444                         pthread_mutex_lock(&evt->mutex);
 445                         remove_watch(listener->watchs);
 446                         pthread_mutex_unlock(&evt->mutex);
 447                 }
 448                 pthread_mutex_unlock(&listener->mutex);
 449
 450                 /* free the listener */
 451                 pthread_mutex_destroy(&listener->mutex);
 452                 free(listener);
 453         }
 454 }
 455
 456 /*
 457  * Makes the 'listener' watching 'event'
 458  * Returns 0 in case of success or else -1.
 459  */
 460 int afb_evt_add_watch(struct afb_evt_listener *listener, struct afb_event event)
 461 {
 462         struct afb_evt_watch *watch;
 463         struct afb_evt_event *evt;
 464
 465         /* check parameter */
 466         if (event.itf != &afb_evt_event_itf || listener->itf->push == NULL) {
 467                 errno = EINVAL;
 468                 return -1;
 469         }
 470
 471         /* search the existing watch for the listener */
 472         evt = event.closure;
 473         pthread_mutex_lock(&listener->mutex);
 474         watch = listener->watchs;
 475         while(watch != NULL) {
 476                 if (watch->event == evt)
 477                         goto found;
 478                 watch = watch->next_by_listener;
 479         }
 480
 481         /* not found, allocate a new */
 482         watch = malloc(sizeof *watch);
 483         if (watch == NULL) {
 484                 pthread_mutex_unlock(&listener->mutex);
 485                 errno = ENOMEM;
 486                 return -1;
 487         }
 488
 489         /* initialise and link */
 490         watch->event = evt;
 491         watch->activity = 0;
 492         watch->listener = listener;
 493         watch->next_by_listener = listener->watchs;
 494         listener->watchs = watch;
 495         pthread_mutex_lock(&evt->mutex);
 496         watch->next_by_event = evt->watchs;
 497         evt->watchs = watch;
 498         pthread_mutex_unlock(&evt->mutex);
 499
 500 found:
 501         if (watch->activity == 0 && listener->itf->add != NULL)
 502                 listener->itf->add(listener->closure, evt->name, evt->id);
 503         watch->activity++;
 504         pthread_mutex_unlock(&listener->mutex);
 505
 506         return 0;
 507 }
 508
 509 /*
 510  * Avoids the 'listener' to watch 'event'
 511  * Returns 0 in case of success or else -1.
 512  */
 513 int afb_evt_remove_watch(struct afb_evt_listener *listener, struct afb_event event)
 514 {
 515         struct afb_evt_watch *watch;
 516         struct afb_evt_event *evt;
 517
 518         /* check parameter */
 519         if (event.itf != &afb_evt_event_itf) {
 520                 errno = EINVAL;
 521                 return -1;
 522         }
 523
 524         /* search the existing watch */
 525         evt = event.closure;
 526         pthread_mutex_lock(&listener->mutex);
 527         watch = listener->watchs;
 528         while(watch != NULL) {
 529                 if (watch->event == evt) {
 530                         if (watch->activity != 0) {
 531                                 watch->activity--;
 532                                 if (watch->activity == 0 && listener->itf->remove != NULL)
 533                                         listener->itf->remove(listener->closure, evt->name, evt->id);
 534                         }
 535                         pthread_mutex_unlock(&listener->mutex);
 536                         return 0;
 537                 }
 538                 watch = watch->next_by_listener;
 539         }
 540         pthread_mutex_unlock(&listener->mutex);
 541         errno = ENOENT;
 542         return -1;
 543 }
 544
 545 /*
 546  * update the hooks for events
 547  */
 548 void afb_evt_update_hooks()
 549 {
 550         struct afb_evt_event *evt;
 551
 552         pthread_mutex_lock(&events_mutex);
 553         for (evt = events ; evt ; evt = evt->next)
 554                 evt->hookflags = afb_hook_flags_evt(evt->name);
 555         pthread_mutex_unlock(&events_mutex);
 556 }
 557
 558 int afb_evt_push(struct afb_event event, struct json_object *object)
 559 {
 560         if (event.itf == &afb_evt_event_itf)
 561                 return evt_push((struct afb_evt_event *)event.closure, object);
 562         json_object_put(object);
 563         return 0;
 564 }
 565
 566 int afb_evt_unhooked_push(struct afb_event event, struct json_object *object)
 567 {
 568         if (event.itf == &afb_evt_event_itf)
 569                 return push((struct afb_evt_event *)event.closure, object, 0);
 570         json_object_put(object);
 571         return 0;
 572 }
 573