src/afb-evt.c

   1 /*
   2  * Copyright (C) 2015-2018 "IoT.bzh"
   3  * Author José Bollo <jose.bollo@iot.bzh>
   4  *
   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
   8  *
   9  *   http://www.apache.org/licenses/LICENSE-2.0
  10  *
  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.
  16  */
  17
  18 #define _GNU_SOURCE
  19
  20 #include <stdlib.h>
  21 #include <string.h>
  22 #include <assert.h>
  23 #include <errno.h>
  24 #include <pthread.h>
  25
  26 #include <json-c/json.h>
  27 #include <afb/afb-event-x2-itf.h>
  28 #include <afb/afb-event-x1.h>
  29
  30 #include "afb-evt.h"
  31 #include "afb-hook.h"
  32 #include "verbose.h"
  33 #include "jobs.h"
  34
  35 struct afb_evt_watch;
  36
  37 /*
  38  * Structure for event listeners
  39  */
  40 struct afb_evt_listener {
  41
  42         /* chaining listeners */
  43         struct afb_evt_listener *next;
  44
  45         /* interface for callbacks */
  46         const struct afb_evt_itf *itf;
  47
  48         /* closure for the callback */
  49         void *closure;
  50
  51         /* head of the list of events listened */
  52         struct afb_evt_watch *watchs;
  53
  54         /* rwlock of the listener */
  55         pthread_rwlock_t rwlock;
  56
  57         /* count of reference to the listener */
  58         int refcount;
  59 };
  60
  61 /*
  62  * Structure for describing events
  63  */
  64 struct afb_evtid {
  65
  66         /* interface */
  67         struct afb_event_x2 eventid;
  68
  69         /* next event */
  70         struct afb_evtid *next;
  71
  72         /* head of the list of listeners watching the event */
  73         struct afb_evt_watch *watchs;
  74
  75         /* rwlock of the event */
  76         pthread_rwlock_t rwlock;
  77
  78         /* hooking */
  79         int hookflags;
  80
  81         /* refcount */
  82         int refcount;
  83
  84         /* id of the event */
  85         int id;
  86
  87         /* has client? */
  88         int has_client;
  89
  90         /* fullname of the event */
  91         char fullname[];
  92 };
  93
  94 /*
  95  * Structure for associating events and listeners
  96  */
  97 struct afb_evt_watch {
  98
  99         /* the evtid */
 100         struct afb_evtid *evtid;
 101
 102         /* link to the next watcher for the same evtid */
 103         struct afb_evt_watch *next_by_evtid;
 104
 105         /* the listener */
 106         struct afb_evt_listener *listener;
 107
 108         /* link to the next watcher for the same listener */
 109         struct afb_evt_watch *next_by_listener;
 110
 111         /* activity */
 112         unsigned activity;
 113 };
 114
 115 /*
 116  * structure for job of broadcasting string events
 117  */
 118 struct job_string
 119 {
 120         /** object atached to the event */
 121         struct json_object *object;
 122
 123         /** name of the event to broadcast */
 124         char event[];
 125 };
 126
 127 /*
 128  * structure for job of broadcasting or pushing events
 129  */
 130 struct job_evtid
 131 {
 132         /** the event to broadcast */
 133         struct afb_evtid *evtid;
 134
 135         /** object atached to the event */
 136         struct json_object *object;
 137 };
 138
 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
 146 };
 147
 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
 155 };
 156
 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)
 160
 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;
 164
 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;
 170
 171 /*
 172  * Create structure for job of broadcasting string 'event' with 'object'
 173  * Returns the created structure or NULL if out of memory
 174  */
 175 static struct job_string *make_job_string(const char *event, struct json_object *object)
 176 {
 177         size_t sz = 1 + strlen(event);
 178         struct job_string *js = malloc(sz + sizeof *js);
 179         if (js) {
 180                 js->object = object;
 181                 memcpy(js->event, event, sz);
 182         }
 183         return js;
 184 }
 185
 186 /*
 187  * Destroy structure 'js' for job of broadcasting string events
 188  */
 189 static void destroy_job_string(struct job_string *js)
 190 {
 191         json_object_put(js->object);
 192         free(js);
 193 }
 194
 195 /*
 196  * Create structure for job of broadcasting or pushing 'evtid' with 'object'
 197  * Returns the created structure or NULL if out of memory
 198  */
 199 static struct job_evtid *make_job_evtid(struct afb_evtid *evtid, struct json_object *object)
 200 {
 201         struct job_evtid *je = malloc(sizeof *je);
 202         if (je) {
 203                 je->evtid = afb_evt_evtid_addref(evtid);
 204                 je->object = object;
 205         }
 206         return je;
 207 }
 208
 209 /*
 210  * Destroy structure for job of broadcasting or pushing evtid
 211  */
 212 static void destroy_job_evtid(struct job_evtid *je)
 213 {
 214         afb_evt_evtid_unref(je->evtid);
 215         json_object_put(je->object);
 216         free(je);
 217 }
 218
 219 /*
 220  * Broadcasts the 'event' of 'id' with its 'object'
 221  */
 222 static void broadcast(const char *event, struct json_object *object, int id)
 223 {
 224         struct afb_evt_listener *listener;
 225
 226         pthread_rwlock_rdlock(&listeners_rwlock);
 227         listener = listeners;
 228         while(listener) {
 229                 if (listener->itf->broadcast != NULL)
 230                         listener->itf->broadcast(listener->closure, event, id, json_object_get(object));
 231                 listener = listener->next;
 232         }
 233         pthread_rwlock_unlock(&listeners_rwlock);
 234 }
 235
 236 /*
 237  * Jobs callback for broadcasting string asynchronously
 238  */
 239 static void broadcast_job_string(int signum, void *closure)
 240 {
 241         struct job_string *js = closure;
 242
 243         if (signum == 0)
 244                 broadcast(js->event, js->object, 0);
 245         destroy_job_string(js);
 246 }
 247
 248 /*
 249  * Jobs callback for broadcasting evtid asynchronously
 250  */
 251 static void broadcast_job_evtid(int signum, void *closure)
 252 {
 253         struct job_evtid *je = closure;
 254
 255         if (signum == 0)
 256                 broadcast(je->evtid->fullname, je->object, je->evtid->id);
 257         destroy_job_evtid(je);
 258 }
 259
 260 /*
 261  * Broadcasts the string 'event' with its 'object'
 262  */
 263 static int broadcast_string(const char *event, struct json_object *object)
 264 {
 265         struct job_string *js;
 266         int rc;
 267
 268         js = make_job_string(event, object);
 269         if (js == NULL) {
 270                 ERROR("Cant't create broadcast string job item for %s(%s)",
 271                         event, json_object_to_json_string(object));
 272                 json_object_put(object);
 273                 return -1;
 274         }
 275
 276         rc = jobs_queue(BROADCAST_JOB_GROUP, 0, broadcast_job_string, js);
 277         if (rc) {
 278                 ERROR("cant't queue broadcast string job item for %s(%s)",
 279                         event, json_object_to_json_string(object));
 280                 destroy_job_string(js);
 281         }
 282         return rc;
 283 }
 284
 285 /*
 286  * Broadcasts the 'evtid' with its 'object'
 287  */
 288 static int broadcast_evtid(struct afb_evtid *evtid, struct json_object *object)
 289 {
 290         struct job_evtid *je;
 291         int rc;
 292
 293         je = make_job_evtid(evtid, object);
 294         if (je == NULL) {
 295                 ERROR("Cant't create broadcast evtid job item for %s(%s)",
 296                         evtid->fullname, json_object_to_json_string(object));
 297                 json_object_put(object);
 298                 return -1;
 299         }
 300
 301         rc = jobs_queue(BROADCAST_JOB_GROUP, 0, broadcast_job_evtid, je);
 302         if (rc) {
 303                 ERROR("cant't queue broadcast evtid job item for %s(%s)",
 304                         evtid->fullname, json_object_to_json_string(object));
 305                 destroy_job_evtid(je);
 306         }
 307         return rc;
 308 }
 309
 310 /*
 311  * Broadcasts the event 'evtid' with its 'object'
 312  * 'object' is released (like json_object_put)
 313  * Returns the count of listener that received the event.
 314  */
 315 int afb_evt_evtid_broadcast(struct afb_evtid *evtid, struct json_object *object)
 316 {
 317         return broadcast_evtid(evtid, object);
 318 }
 319
 320 /*
 321  * Broadcasts the event 'evtid' with its 'object'
 322  * 'object' is released (like json_object_put)
 323  * Returns the count of listener that received the event.
 324  */
 325 int afb_evt_evtid_hooked_broadcast(struct afb_evtid *evtid, struct json_object *object)
 326 {
 327         int result;
 328
 329         json_object_get(object);
 330
 331         if (evtid->hookflags & afb_hook_flag_evt_broadcast_before)
 332                 afb_hook_evt_broadcast_before(evtid->fullname, evtid->id, object);
 333
 334         result = broadcast_evtid(evtid, object);
 335
 336         if (evtid->hookflags & afb_hook_flag_evt_broadcast_after)
 337                 afb_hook_evt_broadcast_after(evtid->fullname, evtid->id, object, result);
 338
 339         json_object_put(object);
 340
 341         return result;
 342 }
 343
 344 /*
 345  * Broadcasts the 'event' with its 'object'
 346  * 'object' is released (like json_object_put)
 347  * Returns the count of listener having receive the event.
 348  */
 349 int afb_evt_broadcast(const char *event, struct json_object *object)
 350 {
 351         int result;
 352
 353         json_object_get(object);
 354
 355         afb_hook_evt_broadcast_before(event, 0, object);
 356         result = broadcast_string(event, object);
 357         afb_hook_evt_broadcast_after(event, 0, object, result);
 358
 359         json_object_put(object);
 360
 361         return result;
 362 }
 363
 364 /*
 365  * Pushes the event 'evtid' with 'obj' to its listeners
 366  * Returns the count of listener that received the event.
 367  */
 368 static void push_evtid(struct afb_evtid *evtid, struct json_object *object)
 369 {
 370         int has_client;
 371         struct afb_evt_watch *watch;
 372         struct afb_evt_listener *listener;
 373
 374         has_client = 0;
 375         pthread_rwlock_rdlock(&evtid->rwlock);
 376         watch = evtid->watchs;
 377         while(watch) {
 378                 listener = watch->listener;
 379                 assert(listener->itf->push != NULL);
 380                 if (watch->activity != 0) {
 381                         listener->itf->push(listener->closure, evtid->fullname, evtid->id, json_object_get(object));
 382                         has_client = 1;
 383                 }
 384                 watch = watch->next_by_evtid;
 385         }
 386         evtid->has_client = has_client;
 387         pthread_rwlock_unlock(&evtid->rwlock);
 388 }
 389
 390 /*
 391  * Jobs callback for pushing evtid asynchronously
 392  */
 393 static void push_job_evtid(int signum, void *closure)
 394 {
 395         struct job_evtid *je = closure;
 396
 397         if (signum == 0)
 398                 push_evtid(je->evtid, je->object);
 399         destroy_job_evtid(je);
 400 }
 401
 402 /*
 403  * Pushes the event 'evtid' with 'obj' to its listeners
 404  * 'obj' is released (like json_object_put)
 405  * Returns 1 if at least one listener exists or 0 if no listener exists or
 406  * -1 in case of error and the event can't be delivered
 407  */
 408 int afb_evt_evtid_push(struct afb_evtid *evtid, struct json_object *object)
 409 {
 410         struct job_evtid *je;
 411         int rc;
 412
 413         je = make_job_evtid(evtid, object);
 414         if (je == NULL) {
 415                 ERROR("Cant't create push evtid job item for %s(%s)",
 416                         evtid->fullname, json_object_to_json_string(object));
 417                 json_object_put(object);
 418                 return -1;
 419         }
 420
 421         rc = jobs_queue(PUSH_JOB_GROUP, 0, push_job_evtid, je);
 422         if (rc == 0)
 423                 rc = evtid->has_client;
 424         else {
 425                 ERROR("cant't queue push evtid job item for %s(%s)",
 426                         evtid->fullname, json_object_to_json_string(object));
 427                 destroy_job_evtid(je);
 428         }
 429
 430         return rc;
 431 }
 432
 433 /*
 434  * Pushes the event 'evtid' with 'obj' to its listeners
 435  * 'obj' is released (like json_object_put)
 436  * Emits calls to hooks.
 437  * Returns the count of listener taht received the event.
 438  */
 439 int afb_evt_evtid_hooked_push(struct afb_evtid *evtid, struct json_object *obj)
 440 {
 441
 442         int result;
 443
 444         /* lease the object */
 445         json_object_get(obj);
 446
 447         /* hook before push */
 448         if (evtid->hookflags & afb_hook_flag_evt_push_before)
 449                 afb_hook_evt_push_before(evtid->fullname, evtid->id, obj);
 450
 451         /* push */
 452         result = afb_evt_evtid_push(evtid, obj);
 453
 454         /* hook after push */
 455         if (evtid->hookflags & afb_hook_flag_evt_push_after)
 456                 afb_hook_evt_push_after(evtid->fullname, evtid->id, obj, result);
 457
 458         /* release the object */
 459         json_object_put(obj);
 460         return result;
 461 }
 462
 463 /*
 464  * remove the 'watch'
 465  */
 466 static void remove_watch(struct afb_evt_watch *watch)
 467 {
 468         struct afb_evt_watch **prv;
 469         struct afb_evtid *evtid;
 470         struct afb_evt_listener *listener;
 471
 472         /* notify listener if needed */
 473         evtid = watch->evtid;
 474         listener = watch->listener;
 475         if (watch->activity != 0 && listener->itf->remove != NULL)
 476                 listener->itf->remove(listener->closure, evtid->fullname, evtid->id);
 477
 478         /* unlink the watch for its event */
 479         prv = &evtid->watchs;
 480         while(*prv != watch)
 481                 prv = &(*prv)->next_by_evtid;
 482         *prv = watch->next_by_evtid;
 483
 484         /* unlink the watch for its listener */
 485         prv = &listener->watchs;
 486         while(*prv != watch)
 487                 prv = &(*prv)->next_by_listener;
 488         *prv = watch->next_by_listener;
 489
 490         /* recycle memory */
 491         free(watch);
 492 }
 493
 494 /*
 495  * Creates an event of name 'fullname' and returns it or NULL on error.
 496  */
 497 struct afb_evtid *afb_evt_evtid_create(const char *fullname)
 498 {
 499         size_t len;
 500         struct afb_evtid *evtid, *oevt;
 501
 502         /* allocates the event */
 503         len = strlen(fullname);
 504         evtid = malloc(len + 1 + sizeof * evtid);
 505         if (evtid == NULL)
 506                 goto error;
 507
 508         /* allocates the id */
 509         pthread_rwlock_wrlock(&events_rwlock);
 510         do {
 511                 if (++event_id_counter < 0) {
 512                         event_id_wrapped = 1;
 513                         event_id_counter = 1024; /* heuristic: small numbers are not destroyed */
 514                 }
 515                 if (!event_id_wrapped)
 516                         break;
 517                 oevt = evtids;
 518                 while(oevt != NULL && oevt->id != event_id_counter)
 519                         oevt = oevt->next;
 520         } while (oevt != NULL);
 521
 522         /* initialize the event */
 523         memcpy(evtid->fullname, fullname, len + 1);
 524         evtid->next = evtids;
 525         evtid->refcount = 1;
 526         evtid->watchs = NULL;
 527         evtid->id = event_id_counter;
 528         evtid->has_client = 0;
 529         pthread_rwlock_init(&evtid->rwlock, NULL);
 530         evtids = evtid;
 531         evtid->hookflags = afb_hook_flags_evt(evtid->fullname);
 532         evtid->eventid.itf = evtid->hookflags ? &afb_evt_hooked_eventid_itf : &afb_evt_event_x2_itf;
 533         if (evtid->hookflags & afb_hook_flag_evt_create)
 534                 afb_hook_evt_create(evtid->fullname, evtid->id);
 535         pthread_rwlock_unlock(&events_rwlock);
 536
 537         /* returns the event */
 538         return evtid;
 539 error:
 540         return NULL;
 541 }
 542
 543 /*
 544  * Creates an event of name 'prefix'/'name' and returns it or NULL on error.
 545  */
 546 struct afb_evtid *afb_evt_evtid_create2(const char *prefix, const char *name)
 547 {
 548         size_t prelen, postlen;
 549         char *fullname;
 550
 551         /* makes the event fullname */
 552         prelen = strlen(prefix);
 553         postlen = strlen(name);
 554         fullname = alloca(prelen + postlen + 2);
 555         memcpy(fullname, prefix, prelen);
 556         fullname[prelen] = '/';
 557         memcpy(fullname + prelen + 1, name, postlen + 1);
 558
 559         /* create the event */
 560         return afb_evt_evtid_create(fullname);
 561 }
 562
 563 /*
 564  * increment the reference count of the event 'evtid'
 565  */
 566 struct afb_evtid *afb_evt_evtid_addref(struct afb_evtid *evtid)
 567 {
 568         __atomic_add_fetch(&evtid->refcount, 1, __ATOMIC_RELAXED);
 569         return evtid;
 570 }
 571
 572 /*
 573  * increment the reference count of the event 'evtid'
 574  */
 575 struct afb_evtid *afb_evt_evtid_hooked_addref(struct afb_evtid *evtid)
 576 {
 577         if (evtid->hookflags & afb_hook_flag_evt_addref)
 578                 afb_hook_evt_addref(evtid->fullname, evtid->id);
 579         return afb_evt_evtid_addref(evtid);
 580 }
 581
 582 /*
 583  * decrement the reference count of the event 'evtid'
 584  * and destroy it when the count reachs zero
 585  */
 586 void afb_evt_evtid_unref(struct afb_evtid *evtid)
 587 {
 588         int found;
 589         struct afb_evtid **prv;
 590         struct afb_evt_listener *listener;
 591
 592         if (!__atomic_sub_fetch(&evtid->refcount, 1, __ATOMIC_RELAXED)) {
 593                 /* unlinks the event if valid! */
 594                 pthread_rwlock_wrlock(&events_rwlock);
 595                 found = 0;
 596                 prv = &evtids;
 597                 while (*prv && !(found = (*prv == evtid)))
 598                         prv = &(*prv)->next;
 599                 if (found)
 600                         *prv = evtid->next;
 601                 pthread_rwlock_unlock(&events_rwlock);
 602
 603                 /* destroys the event */
 604                 if (!found)
 605                         ERROR("event not found");
 606                 else {
 607                         /* removes all watchers */
 608                         while(evtid->watchs != NULL) {
 609                                 listener = evtid->watchs->listener;
 610                                 pthread_rwlock_wrlock(&listener->rwlock);
 611                                 pthread_rwlock_wrlock(&evtid->rwlock);
 612                                 remove_watch(evtid->watchs);
 613                                 pthread_rwlock_unlock(&evtid->rwlock);
 614                                 pthread_rwlock_unlock(&listener->rwlock);
 615                         }
 616
 617                         /* free */
 618                         pthread_rwlock_destroy(&evtid->rwlock);
 619                         free(evtid);
 620                 }
 621         }
 622 }
 623
 624 /*
 625  * decrement the reference count of the event 'evtid'
 626  * and destroy it when the count reachs zero
 627  */
 628 void afb_evt_evtid_hooked_unref(struct afb_evtid *evtid)
 629 {
 630         if (evtid->hookflags & afb_hook_flag_evt_unref)
 631                 afb_hook_evt_unref(evtid->fullname, evtid->id);
 632         afb_evt_evtid_unref(evtid);
 633 }
 634
 635 /*
 636  * Returns the true name of the 'event'
 637  */
 638 const char *afb_evt_evtid_fullname(struct afb_evtid *evtid)
 639 {
 640         return evtid->fullname;
 641 }
 642
 643 /*
 644  * Returns the name of the 'event'
 645  */
 646 const char *afb_evt_evtid_name(struct afb_evtid *evtid)
 647 {
 648         const char *name = strchr(evtid->fullname, '/');
 649         return name ? name + 1 : evtid->fullname;
 650 }
 651
 652 /*
 653  * Returns the name associated to the event 'evtid'.
 654  */
 655 const char *afb_evt_evtid_hooked_name(struct afb_evtid *evtid)
 656 {
 657         const char *result = afb_evt_evtid_name(evtid);
 658         if (evtid->hookflags & afb_hook_flag_evt_name)
 659                 afb_hook_evt_name(evtid->fullname, evtid->id, result);
 660         return result;
 661 }
 662
 663 /*
 664  * Returns the id of the 'event'
 665  */
 666 int afb_evt_evtid_id(struct afb_evtid *evtid)
 667 {
 668         return evtid->id;
 669 }
 670
 671 /*
 672  * Returns an instance of the listener defined by the 'send' callback
 673  * and the 'closure'.
 674  * Returns NULL in case of memory depletion.
 675  */
 676 struct afb_evt_listener *afb_evt_listener_create(const struct afb_evt_itf *itf, void *closure)
 677 {
 678         struct afb_evt_listener *listener;
 679
 680         /* search if an instance already exists */
 681         pthread_rwlock_wrlock(&listeners_rwlock);
 682         listener = listeners;
 683         while (listener != NULL) {
 684                 if (listener->itf == itf && listener->closure == closure) {
 685                         listener = afb_evt_listener_addref(listener);
 686                         goto found;
 687                 }
 688                 listener = listener->next;
 689         }
 690
 691         /* allocates */
 692         listener = calloc(1, sizeof *listener);
 693         if (listener != NULL) {
 694                 /* init */
 695                 listener->itf = itf;
 696                 listener->closure = closure;
 697                 listener->watchs = NULL;
 698                 listener->refcount = 1;
 699                 pthread_rwlock_init(&listener->rwlock, NULL);
 700                 listener->next = listeners;
 701                 listeners = listener;
 702         }
 703  found:
 704         pthread_rwlock_unlock(&listeners_rwlock);
 705         return listener;
 706 }
 707
 708 /*
 709  * Increases the reference count of 'listener' and returns it
 710  */
 711 struct afb_evt_listener *afb_evt_listener_addref(struct afb_evt_listener *listener)
 712 {
 713         __atomic_add_fetch(&listener->refcount, 1, __ATOMIC_RELAXED);
 714         return listener;
 715 }
 716
 717 /*
 718  * Decreases the reference count of the 'listener' and destroys it
 719  * when no more used.
 720  */
 721 void afb_evt_listener_unref(struct afb_evt_listener *listener)
 722 {
 723         struct afb_evt_listener **prv;
 724         struct afb_evtid *evtid;
 725
 726         if (listener && !__atomic_sub_fetch(&listener->refcount, 1, __ATOMIC_RELAXED)) {
 727
 728                 /* unlink the listener */
 729                 pthread_rwlock_wrlock(&listeners_rwlock);
 730                 prv = &listeners;
 731                 while (*prv != listener)
 732                         prv = &(*prv)->next;
 733                 *prv = listener->next;
 734                 pthread_rwlock_unlock(&listeners_rwlock);
 735
 736                 /* remove the watchers */
 737                 pthread_rwlock_wrlock(&listener->rwlock);
 738                 while (listener->watchs != NULL) {
 739                         evtid = listener->watchs->evtid;
 740                         pthread_rwlock_wrlock(&evtid->rwlock);
 741                         remove_watch(listener->watchs);
 742                         pthread_rwlock_unlock(&evtid->rwlock);
 743                 }
 744                 pthread_rwlock_unlock(&listener->rwlock);
 745
 746                 /* free the listener */
 747                 pthread_rwlock_destroy(&listener->rwlock);
 748                 free(listener);
 749         }
 750 }
 751
 752 /*
 753  * Makes the 'listener' watching 'evtid'
 754  * Returns 0 in case of success or else -1.
 755  */
 756 int afb_evt_watch_add_evtid(struct afb_evt_listener *listener, struct afb_evtid *evtid)
 757 {
 758         struct afb_evt_watch *watch;
 759
 760         /* check parameter */
 761         if (listener->itf->push == NULL) {
 762                 errno = EINVAL;
 763                 return -1;
 764         }
 765
 766         /* search the existing watch for the listener */
 767         pthread_rwlock_wrlock(&listener->rwlock);
 768         watch = listener->watchs;
 769         while(watch != NULL) {
 770                 if (watch->evtid == evtid)
 771                         goto found;
 772                 watch = watch->next_by_listener;
 773         }
 774
 775         /* not found, allocate a new */
 776         watch = malloc(sizeof *watch);
 777         if (watch == NULL) {
 778                 pthread_rwlock_unlock(&listener->rwlock);
 779                 errno = ENOMEM;
 780                 return -1;
 781         }
 782
 783         /* initialise and link */
 784         watch->evtid = evtid;
 785         watch->activity = 0;
 786         watch->listener = listener;
 787         watch->next_by_listener = listener->watchs;
 788         listener->watchs = watch;
 789         pthread_rwlock_wrlock(&evtid->rwlock);
 790         watch->next_by_evtid = evtid->watchs;
 791         evtid->watchs = watch;
 792         pthread_rwlock_unlock(&evtid->rwlock);
 793
 794 found:
 795         if (watch->activity == 0 && listener->itf->add != NULL)
 796                 listener->itf->add(listener->closure, evtid->fullname, evtid->id);
 797         watch->activity++;
 798         evtid->has_client = 1;
 799         pthread_rwlock_unlock(&listener->rwlock);
 800
 801         return 0;
 802 }
 803
 804 /*
 805  * Avoids the 'listener' to watch 'evtid'
 806  * Returns 0 in case of success or else -1.
 807  */
 808 int afb_evt_watch_sub_evtid(struct afb_evt_listener *listener, struct afb_evtid *evtid)
 809 {
 810         struct afb_evt_watch *watch;
 811
 812         /* search the existing watch */
 813         pthread_rwlock_wrlock(&listener->rwlock);
 814         watch = listener->watchs;
 815         while(watch != NULL) {
 816                 if (watch->evtid == evtid) {
 817                         if (watch->activity != 0) {
 818                                 watch->activity--;
 819                                 if (watch->activity == 0 && listener->itf->remove != NULL)
 820                                         listener->itf->remove(listener->closure, evtid->fullname, evtid->id);
 821                         }
 822                         pthread_rwlock_unlock(&listener->rwlock);
 823                         return 0;
 824                 }
 825                 watch = watch->next_by_listener;
 826         }
 827         pthread_rwlock_unlock(&listener->rwlock);
 828         errno = ENOENT;
 829         return -1;
 830 }
 831
 832 /*
 833  * update the hooks for events
 834  */
 835 void afb_evt_update_hooks()
 836 {
 837         struct afb_evtid *evtid;
 838
 839         pthread_rwlock_rdlock(&events_rwlock);
 840         for (evtid = evtids ; evtid ; evtid = evtid->next) {
 841                 evtid->hookflags = afb_hook_flags_evt(evtid->fullname);
 842                 evtid->eventid.itf = evtid->hookflags ? &afb_evt_hooked_eventid_itf : &afb_evt_event_x2_itf;
 843         }
 844         pthread_rwlock_unlock(&events_rwlock);
 845 }
 846
 847 inline struct afb_evtid *afb_evt_event_x2_to_evtid(struct afb_event_x2 *eventid)
 848 {
 849         return (struct afb_evtid*)eventid;
 850 }
 851
 852 inline struct afb_event_x2 *afb_evt_event_x2_from_evtid(struct afb_evtid *evtid)
 853 {
 854         return &evtid->eventid;
 855 }
 856
 857 /*
 858  * Creates an event of 'fullname' and returns it.
 859  * Returns an event with closure==NULL in case of error.
 860  */
 861 struct afb_event_x2 *afb_evt_event_x2_create(const char *fullname)
 862 {
 863         return afb_evt_event_x2_from_evtid(afb_evt_evtid_create(fullname));
 864 }
 865
 866 /*
 867  * Creates an event of name 'prefix'/'name' and returns it.
 868  * Returns an event with closure==NULL in case of error.
 869  */
 870 struct afb_event_x2 *afb_evt_event_x2_create2(const char *prefix, const char *name)
 871 {
 872         return afb_evt_event_x2_from_evtid(afb_evt_evtid_create2(prefix, name));
 873 }
 874
 875 /*
 876  * Returns the fullname of the 'eventid'
 877  */
 878 const char *afb_evt_event_x2_fullname(struct afb_event_x2 *eventid)
 879 {
 880         struct afb_evtid *evtid = afb_evt_event_x2_to_evtid(eventid);
 881         return evtid ? evtid->fullname : NULL;
 882 }
 883
 884 /*
 885  * Returns the id of the 'eventid'
 886  */
 887 int afb_evt_event_x2_id(struct afb_event_x2 *eventid)
 888 {
 889         struct afb_evtid *evtid = afb_evt_event_x2_to_evtid(eventid);
 890         return evtid ? evtid->id : 0;
 891 }
 892
 893 /*
 894  * Makes the 'listener' watching 'eventid'
 895  * Returns 0 in case of success or else -1.
 896  */
 897 int afb_evt_event_x2_add_watch(struct afb_evt_listener *listener, struct afb_event_x2 *eventid)
 898 {
 899         struct afb_evtid *evtid = afb_evt_event_x2_to_evtid(eventid);
 900
 901         /* check parameter */
 902         if (!evtid) {
 903                 errno = EINVAL;
 904                 return -1;
 905         }
 906
 907         /* search the existing watch for the listener */
 908         return afb_evt_watch_add_evtid(listener, evtid);
 909 }
 910
 911 /*
 912  * Avoids the 'listener' to watch 'eventid'
 913  * Returns 0 in case of success or else -1.
 914  */
 915 int afb_evt_event_x2_remove_watch(struct afb_evt_listener *listener, struct afb_event_x2 *eventid)
 916 {
 917         struct afb_evtid *evtid = afb_evt_event_x2_to_evtid(eventid);
 918
 919         /* check parameter */
 920         if (!evtid) {
 921                 errno = EINVAL;
 922                 return -1;
 923         }
 924
 925         /* search the existing watch */
 926         return afb_evt_watch_sub_evtid(listener, evtid);
 927 }
 928
 929 int afb_evt_event_x2_push(struct afb_event_x2 *eventid, struct json_object *object)
 930 {
 931         struct afb_evtid *evtid = afb_evt_event_x2_to_evtid(eventid);
 932         if (evtid)
 933                 return afb_evt_evtid_hooked_push(evtid, object);
 934         json_object_put(object);
 935         return 0;
 936 }
 937
 938 int afb_evt_event_x2_unhooked_push(struct afb_event_x2 *eventid, struct json_object *object)
 939 {
 940         struct afb_evtid *evtid = afb_evt_event_x2_to_evtid(eventid);
 941         if (evtid)
 942                 return afb_evt_evtid_push(evtid, object);
 943         json_object_put(object);
 944         return 0;
 945 }
 946
 947 struct afb_event_x1 afb_evt_event_from_evtid(struct afb_evtid *evtid)
 948 {
 949         return evtid
 950                 ? (struct afb_event_x1){ .itf = &afb_evt_hooked_eventid_itf, .closure = &evtid->eventid }
 951                 : (struct afb_event_x1){ .itf = NULL, .closure = NULL };
 952 }
 953
 954 void afb_evt_event_x2_unref(struct afb_event_x2 *eventid)
 955 {
 956         struct afb_evtid *evtid = afb_evt_event_x2_to_evtid(eventid);
 957         if (evtid)
 958                 afb_evt_evtid_unref(evtid);
 959 }
 960
 961 struct afb_event_x2 *afb_evt_event_x2_addref(struct afb_event_x2 *eventid)
 962 {
 963         struct afb_evtid *evtid = afb_evt_event_x2_to_evtid(eventid);
 964         if (evtid)
 965                 afb_evt_evtid_addref(evtid);
 966         return eventid;
 967 }
 968