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 events
 117  */
 118 struct job_broadcast
 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_broadcast *make_job_broadcast(const char *event, struct json_object *object)
 176 {
 177         size_t sz = 1 + strlen(event);
 178         struct job_broadcast *jb = malloc(sz + sizeof *jb);
 179         if (jb) {
 180                 jb->object = object;
 181                 memcpy(jb->event, event, sz);
 182         }
 183         return jb;
 184 }
 185
 186 /*
 187  * Destroy structure 'jb' for job of broadcasting string events
 188  */
 189 static void destroy_job_broadcast(struct job_broadcast *jb)
 190 {
 191         json_object_put(jb->object);
 192         free(jb);
 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)
 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, 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(int signum, void *closure)
 240 {
 241         struct job_broadcast *jb = closure;
 242
 243         if (signum == 0)
 244                 broadcast(jb->event, jb->object);
 245         destroy_job_broadcast(jb);
 246 }
 247
 248 /*
 249  * Broadcasts the string 'event' with its 'object'
 250  */
 251 static int unhooked_broadcast(const char *event, struct json_object *object)
 252 {
 253         struct job_broadcast *jb;
 254         int rc;
 255
 256         jb = make_job_broadcast(event, object);
 257         if (jb == NULL) {
 258                 ERROR("Cant't create broadcast string job item for %s(%s)",
 259                         event, json_object_to_json_string(object));
 260                 json_object_put(object);
 261                 return -1;
 262         }
 263
 264         rc = jobs_queue(BROADCAST_JOB_GROUP, 0, broadcast_job, jb);
 265         if (rc) {
 266                 ERROR("cant't queue broadcast string job item for %s(%s)",
 267                         event, json_object_to_json_string(object));
 268                 destroy_job_broadcast(jb);
 269         }
 270         return rc;
 271 }
 272
 273 /*
 274  * Broadcasts the event 'evtid' with its 'object'
 275  * 'object' is released (like json_object_put)
 276  * Returns the count of listener that received the event.
 277  */
 278 int afb_evt_evtid_broadcast(struct afb_evtid *evtid, struct json_object *object)
 279 {
 280         return unhooked_broadcast(evtid->fullname, object);
 281 }
 282
 283 /*
 284  * Broadcasts the event 'evtid' with its 'object'
 285  * 'object' is released (like json_object_put)
 286  * Returns the count of listener that received the event.
 287  */
 288 int afb_evt_evtid_hooked_broadcast(struct afb_evtid *evtid, struct json_object *object)
 289 {
 290         int result;
 291
 292         json_object_get(object);
 293
 294         if (evtid->hookflags & afb_hook_flag_evt_broadcast_before)
 295                 afb_hook_evt_broadcast_before(evtid->fullname, evtid->id, object);
 296
 297         result = afb_evt_evtid_broadcast(evtid, object);
 298
 299         if (evtid->hookflags & afb_hook_flag_evt_broadcast_after)
 300                 afb_hook_evt_broadcast_after(evtid->fullname, evtid->id, object, result);
 301
 302         json_object_put(object);
 303
 304         return result;
 305 }
 306
 307 /*
 308  * Broadcasts the 'event' with its 'object'
 309  * 'object' is released (like json_object_put)
 310  * Returns the count of listener having receive the event.
 311  */
 312 int afb_evt_broadcast(const char *event, struct json_object *object)
 313 {
 314         int result;
 315
 316 #if WITH_AFB_HOOK
 317         json_object_get(object);
 318         afb_hook_evt_broadcast_before(event, 0, object);
 319 #endif
 320
 321         result = unhooked_broadcast(event, object);
 322
 323 #if WITH_AFB_HOOK
 324         afb_hook_evt_broadcast_after(event, 0, object, result);
 325         json_object_put(object);
 326 #endif
 327         return result;
 328 }
 329
 330 /*
 331  * Pushes the event 'evtid' with 'obj' to its listeners
 332  * Returns the count of listener that received the event.
 333  */
 334 static void push_evtid(struct afb_evtid *evtid, struct json_object *object)
 335 {
 336         int has_client;
 337         struct afb_evt_watch *watch;
 338         struct afb_evt_listener *listener;
 339
 340         has_client = 0;
 341         pthread_rwlock_rdlock(&evtid->rwlock);
 342         watch = evtid->watchs;
 343         while(watch) {
 344                 listener = watch->listener;
 345                 assert(listener->itf->push != NULL);
 346                 if (watch->activity != 0) {
 347                         listener->itf->push(listener->closure, evtid->fullname, evtid->id, json_object_get(object));
 348                         has_client = 1;
 349                 }
 350                 watch = watch->next_by_evtid;
 351         }
 352         evtid->has_client = has_client;
 353         pthread_rwlock_unlock(&evtid->rwlock);
 354 }
 355
 356 /*
 357  * Jobs callback for pushing evtid asynchronously
 358  */
 359 static void push_job_evtid(int signum, void *closure)
 360 {
 361         struct job_evtid *je = closure;
 362
 363         if (signum == 0)
 364                 push_evtid(je->evtid, je->object);
 365         destroy_job_evtid(je);
 366 }
 367
 368 /*
 369  * Pushes the event 'evtid' with 'obj' to its listeners
 370  * 'obj' is released (like json_object_put)
 371  * Returns 1 if at least one listener exists or 0 if no listener exists or
 372  * -1 in case of error and the event can't be delivered
 373  */
 374 int afb_evt_evtid_push(struct afb_evtid *evtid, struct json_object *object)
 375 {
 376         struct job_evtid *je;
 377         int rc;
 378
 379         je = make_job_evtid(evtid, object);
 380         if (je == NULL) {
 381                 ERROR("Cant't create push evtid job item for %s(%s)",
 382                         evtid->fullname, json_object_to_json_string(object));
 383                 json_object_put(object);
 384                 return -1;
 385         }
 386
 387         rc = jobs_queue(PUSH_JOB_GROUP, 0, push_job_evtid, je);
 388         if (rc == 0)
 389                 rc = evtid->has_client;
 390         else {
 391                 ERROR("cant't queue push evtid job item for %s(%s)",
 392                         evtid->fullname, json_object_to_json_string(object));
 393                 destroy_job_evtid(je);
 394         }
 395
 396         return rc;
 397 }
 398
 399 /*
 400  * Pushes the event 'evtid' with 'obj' to its listeners
 401  * 'obj' is released (like json_object_put)
 402  * Emits calls to hooks.
 403  * Returns the count of listener taht received the event.
 404  */
 405 int afb_evt_evtid_hooked_push(struct afb_evtid *evtid, struct json_object *obj)
 406 {
 407
 408         int result;
 409
 410         /* lease the object */
 411         json_object_get(obj);
 412
 413         /* hook before push */
 414         if (evtid->hookflags & afb_hook_flag_evt_push_before)
 415                 afb_hook_evt_push_before(evtid->fullname, evtid->id, obj);
 416
 417         /* push */
 418         result = afb_evt_evtid_push(evtid, obj);
 419
 420         /* hook after push */
 421         if (evtid->hookflags & afb_hook_flag_evt_push_after)
 422                 afb_hook_evt_push_after(evtid->fullname, evtid->id, obj, result);
 423
 424         /* release the object */
 425         json_object_put(obj);
 426         return result;
 427 }
 428
 429 /*
 430  * remove the 'watch'
 431  */
 432 static void remove_watch(struct afb_evt_watch *watch)
 433 {
 434         struct afb_evt_watch **prv;
 435         struct afb_evtid *evtid;
 436         struct afb_evt_listener *listener;
 437
 438         /* notify listener if needed */
 439         evtid = watch->evtid;
 440         listener = watch->listener;
 441         if (watch->activity != 0 && listener->itf->remove != NULL)
 442                 listener->itf->remove(listener->closure, evtid->fullname, evtid->id);
 443
 444         /* unlink the watch for its event */
 445         prv = &evtid->watchs;
 446         while(*prv != watch)
 447                 prv = &(*prv)->next_by_evtid;
 448         *prv = watch->next_by_evtid;
 449
 450         /* unlink the watch for its listener */
 451         prv = &listener->watchs;
 452         while(*prv != watch)
 453                 prv = &(*prv)->next_by_listener;
 454         *prv = watch->next_by_listener;
 455
 456         /* recycle memory */
 457         free(watch);
 458 }
 459
 460 /*
 461  * Creates an event of name 'fullname' and returns it or NULL on error.
 462  */
 463 struct afb_evtid *afb_evt_evtid_create(const char *fullname)
 464 {
 465         size_t len;
 466         struct afb_evtid *evtid, *oevt;
 467
 468         /* allocates the event */
 469         len = strlen(fullname);
 470         evtid = malloc(len + 1 + sizeof * evtid);
 471         if (evtid == NULL)
 472                 goto error;
 473
 474         /* allocates the id */
 475         pthread_rwlock_wrlock(&events_rwlock);
 476         do {
 477                 if (++event_id_counter < 0) {
 478                         event_id_wrapped = 1;
 479                         event_id_counter = 1024; /* heuristic: small numbers are not destroyed */
 480                 }
 481                 if (!event_id_wrapped)
 482                         break;
 483                 oevt = evtids;
 484                 while(oevt != NULL && oevt->id != event_id_counter)
 485                         oevt = oevt->next;
 486         } while (oevt != NULL);
 487
 488         /* initialize the event */
 489         memcpy(evtid->fullname, fullname, len + 1);
 490         evtid->next = evtids;
 491         evtid->refcount = 1;
 492         evtid->watchs = NULL;
 493         evtid->id = event_id_counter;
 494         evtid->has_client = 0;
 495         pthread_rwlock_init(&evtid->rwlock, NULL);
 496         evtids = evtid;
 497         evtid->hookflags = afb_hook_flags_evt(evtid->fullname);
 498         evtid->eventid.itf = evtid->hookflags ? &afb_evt_hooked_eventid_itf : &afb_evt_event_x2_itf;
 499         if (evtid->hookflags & afb_hook_flag_evt_create)
 500                 afb_hook_evt_create(evtid->fullname, evtid->id);
 501         pthread_rwlock_unlock(&events_rwlock);
 502
 503         /* returns the event */
 504         return evtid;
 505 error:
 506         return NULL;
 507 }
 508
 509 /*
 510  * Creates an event of name 'prefix'/'name' and returns it or NULL on error.
 511  */
 512 struct afb_evtid *afb_evt_evtid_create2(const char *prefix, const char *name)
 513 {
 514         size_t prelen, postlen;
 515         char *fullname;
 516
 517         /* makes the event fullname */
 518         prelen = strlen(prefix);
 519         postlen = strlen(name);
 520         fullname = alloca(prelen + postlen + 2);
 521         memcpy(fullname, prefix, prelen);
 522         fullname[prelen] = '/';
 523         memcpy(fullname + prelen + 1, name, postlen + 1);
 524
 525         /* create the event */
 526         return afb_evt_evtid_create(fullname);
 527 }
 528
 529 /*
 530  * increment the reference count of the event 'evtid'
 531  */
 532 struct afb_evtid *afb_evt_evtid_addref(struct afb_evtid *evtid)
 533 {
 534         __atomic_add_fetch(&evtid->refcount, 1, __ATOMIC_RELAXED);
 535         return evtid;
 536 }
 537
 538 /*
 539  * increment the reference count of the event 'evtid'
 540  */
 541 struct afb_evtid *afb_evt_evtid_hooked_addref(struct afb_evtid *evtid)
 542 {
 543         if (evtid->hookflags & afb_hook_flag_evt_addref)
 544                 afb_hook_evt_addref(evtid->fullname, evtid->id);
 545         return afb_evt_evtid_addref(evtid);
 546 }
 547
 548 /*
 549  * decrement the reference count of the event 'evtid'
 550  * and destroy it when the count reachs zero
 551  */
 552 void afb_evt_evtid_unref(struct afb_evtid *evtid)
 553 {
 554         int found;
 555         struct afb_evtid **prv;
 556         struct afb_evt_listener *listener;
 557
 558         if (!__atomic_sub_fetch(&evtid->refcount, 1, __ATOMIC_RELAXED)) {
 559                 /* unlinks the event if valid! */
 560                 pthread_rwlock_wrlock(&events_rwlock);
 561                 found = 0;
 562                 prv = &evtids;
 563                 while (*prv && !(found = (*prv == evtid)))
 564                         prv = &(*prv)->next;
 565                 if (found)
 566                         *prv = evtid->next;
 567                 pthread_rwlock_unlock(&events_rwlock);
 568
 569                 /* destroys the event */
 570                 if (!found)
 571                         ERROR("event not found");
 572                 else {
 573                         /* removes all watchers */
 574                         while(evtid->watchs != NULL) {
 575                                 listener = evtid->watchs->listener;
 576                                 pthread_rwlock_wrlock(&listener->rwlock);
 577                                 pthread_rwlock_wrlock(&evtid->rwlock);
 578                                 remove_watch(evtid->watchs);
 579                                 pthread_rwlock_unlock(&evtid->rwlock);
 580                                 pthread_rwlock_unlock(&listener->rwlock);
 581                         }
 582
 583                         /* free */
 584                         pthread_rwlock_destroy(&evtid->rwlock);
 585                         free(evtid);
 586                 }
 587         }
 588 }
 589
 590 /*
 591  * decrement the reference count of the event 'evtid'
 592  * and destroy it when the count reachs zero
 593  */
 594 void afb_evt_evtid_hooked_unref(struct afb_evtid *evtid)
 595 {
 596         if (evtid->hookflags & afb_hook_flag_evt_unref)
 597                 afb_hook_evt_unref(evtid->fullname, evtid->id);
 598         afb_evt_evtid_unref(evtid);
 599 }
 600
 601 /*
 602  * Returns the true name of the 'event'
 603  */
 604 const char *afb_evt_evtid_fullname(struct afb_evtid *evtid)
 605 {
 606         return evtid->fullname;
 607 }
 608
 609 /*
 610  * Returns the name of the 'event'
 611  */
 612 const char *afb_evt_evtid_name(struct afb_evtid *evtid)
 613 {
 614         const char *name = strchr(evtid->fullname, '/');
 615         return name ? name + 1 : evtid->fullname;
 616 }
 617
 618 /*
 619  * Returns the name associated to the event 'evtid'.
 620  */
 621 const char *afb_evt_evtid_hooked_name(struct afb_evtid *evtid)
 622 {
 623         const char *result = afb_evt_evtid_name(evtid);
 624         if (evtid->hookflags & afb_hook_flag_evt_name)
 625                 afb_hook_evt_name(evtid->fullname, evtid->id, result);
 626         return result;
 627 }
 628
 629 /*
 630  * Returns the id of the 'event'
 631  */
 632 int afb_evt_evtid_id(struct afb_evtid *evtid)
 633 {
 634         return evtid->id;
 635 }
 636
 637 /*
 638  * Returns an instance of the listener defined by the 'send' callback
 639  * and the 'closure'.
 640  * Returns NULL in case of memory depletion.
 641  */
 642 struct afb_evt_listener *afb_evt_listener_create(const struct afb_evt_itf *itf, void *closure)
 643 {
 644         struct afb_evt_listener *listener;
 645
 646         /* search if an instance already exists */
 647         pthread_rwlock_wrlock(&listeners_rwlock);
 648         listener = listeners;
 649         while (listener != NULL) {
 650                 if (listener->itf == itf && listener->closure == closure) {
 651                         listener = afb_evt_listener_addref(listener);
 652                         goto found;
 653                 }
 654                 listener = listener->next;
 655         }
 656
 657         /* allocates */
 658         listener = calloc(1, sizeof *listener);
 659         if (listener != NULL) {
 660                 /* init */
 661                 listener->itf = itf;
 662                 listener->closure = closure;
 663                 listener->watchs = NULL;
 664                 listener->refcount = 1;
 665                 pthread_rwlock_init(&listener->rwlock, NULL);
 666                 listener->next = listeners;
 667                 listeners = listener;
 668         }
 669  found:
 670         pthread_rwlock_unlock(&listeners_rwlock);
 671         return listener;
 672 }
 673
 674 /*
 675  * Increases the reference count of 'listener' and returns it
 676  */
 677 struct afb_evt_listener *afb_evt_listener_addref(struct afb_evt_listener *listener)
 678 {
 679         __atomic_add_fetch(&listener->refcount, 1, __ATOMIC_RELAXED);
 680         return listener;
 681 }
 682
 683 /*
 684  * Decreases the reference count of the 'listener' and destroys it
 685  * when no more used.
 686  */
 687 void afb_evt_listener_unref(struct afb_evt_listener *listener)
 688 {
 689         struct afb_evt_listener **prv;
 690         struct afb_evtid *evtid;
 691
 692         if (listener && !__atomic_sub_fetch(&listener->refcount, 1, __ATOMIC_RELAXED)) {
 693
 694                 /* unlink the listener */
 695                 pthread_rwlock_wrlock(&listeners_rwlock);
 696                 prv = &listeners;
 697                 while (*prv != listener)
 698                         prv = &(*prv)->next;
 699                 *prv = listener->next;
 700                 pthread_rwlock_unlock(&listeners_rwlock);
 701
 702                 /* remove the watchers */
 703                 pthread_rwlock_wrlock(&listener->rwlock);
 704                 while (listener->watchs != NULL) {
 705                         evtid = listener->watchs->evtid;
 706                         pthread_rwlock_wrlock(&evtid->rwlock);
 707                         remove_watch(listener->watchs);
 708                         pthread_rwlock_unlock(&evtid->rwlock);
 709                 }
 710                 pthread_rwlock_unlock(&listener->rwlock);
 711
 712                 /* free the listener */
 713                 pthread_rwlock_destroy(&listener->rwlock);
 714                 free(listener);
 715         }
 716 }
 717
 718 /*
 719  * Makes the 'listener' watching 'evtid'
 720  * Returns 0 in case of success or else -1.
 721  */
 722 int afb_evt_watch_add_evtid(struct afb_evt_listener *listener, struct afb_evtid *evtid)
 723 {
 724         struct afb_evt_watch *watch;
 725
 726         /* check parameter */
 727         if (listener->itf->push == NULL) {
 728                 errno = EINVAL;
 729                 return -1;
 730         }
 731
 732         /* search the existing watch for the listener */
 733         pthread_rwlock_wrlock(&listener->rwlock);
 734         watch = listener->watchs;
 735         while(watch != NULL) {
 736                 if (watch->evtid == evtid)
 737                         goto found;
 738                 watch = watch->next_by_listener;
 739         }
 740
 741         /* not found, allocate a new */
 742         watch = malloc(sizeof *watch);
 743         if (watch == NULL) {
 744                 pthread_rwlock_unlock(&listener->rwlock);
 745                 errno = ENOMEM;
 746                 return -1;
 747         }
 748
 749         /* initialise and link */
 750         watch->evtid = evtid;
 751         watch->activity = 0;
 752         watch->listener = listener;
 753         watch->next_by_listener = listener->watchs;
 754         listener->watchs = watch;
 755         pthread_rwlock_wrlock(&evtid->rwlock);
 756         watch->next_by_evtid = evtid->watchs;
 757         evtid->watchs = watch;
 758         pthread_rwlock_unlock(&evtid->rwlock);
 759
 760 found:
 761         if (watch->activity == 0 && listener->itf->add != NULL)
 762                 listener->itf->add(listener->closure, evtid->fullname, evtid->id);
 763         watch->activity++;
 764         evtid->has_client = 1;
 765         pthread_rwlock_unlock(&listener->rwlock);
 766
 767         return 0;
 768 }
 769
 770 /*
 771  * Avoids the 'listener' to watch 'evtid'
 772  * Returns 0 in case of success or else -1.
 773  */
 774 int afb_evt_watch_sub_evtid(struct afb_evt_listener *listener, struct afb_evtid *evtid)
 775 {
 776         struct afb_evt_watch *watch;
 777
 778         /* search the existing watch */
 779         pthread_rwlock_wrlock(&listener->rwlock);
 780         watch = listener->watchs;
 781         while(watch != NULL) {
 782                 if (watch->evtid == evtid) {
 783                         if (watch->activity != 0) {
 784                                 watch->activity--;
 785                                 if (watch->activity == 0 && listener->itf->remove != NULL)
 786                                         listener->itf->remove(listener->closure, evtid->fullname, evtid->id);
 787                         }
 788                         pthread_rwlock_unlock(&listener->rwlock);
 789                         return 0;
 790                 }
 791                 watch = watch->next_by_listener;
 792         }
 793         pthread_rwlock_unlock(&listener->rwlock);
 794         errno = ENOENT;
 795         return -1;
 796 }
 797
 798 /*
 799  * update the hooks for events
 800  */
 801 void afb_evt_update_hooks()
 802 {
 803         struct afb_evtid *evtid;
 804
 805         pthread_rwlock_rdlock(&events_rwlock);
 806         for (evtid = evtids ; evtid ; evtid = evtid->next) {
 807                 evtid->hookflags = afb_hook_flags_evt(evtid->fullname);
 808                 evtid->eventid.itf = evtid->hookflags ? &afb_evt_hooked_eventid_itf : &afb_evt_event_x2_itf;
 809         }
 810         pthread_rwlock_unlock(&events_rwlock);
 811 }
 812
 813 inline struct afb_evtid *afb_evt_event_x2_to_evtid(struct afb_event_x2 *eventid)
 814 {
 815         return (struct afb_evtid*)eventid;
 816 }
 817
 818 inline struct afb_event_x2 *afb_evt_event_x2_from_evtid(struct afb_evtid *evtid)
 819 {
 820         return &evtid->eventid;
 821 }
 822
 823 /*
 824  * Creates an event of 'fullname' and returns it.
 825  * Returns an event with closure==NULL in case of error.
 826  */
 827 struct afb_event_x2 *afb_evt_event_x2_create(const char *fullname)
 828 {
 829         return afb_evt_event_x2_from_evtid(afb_evt_evtid_create(fullname));
 830 }
 831
 832 /*
 833  * Creates an event of name 'prefix'/'name' and returns it.
 834  * Returns an event with closure==NULL in case of error.
 835  */
 836 struct afb_event_x2 *afb_evt_event_x2_create2(const char *prefix, const char *name)
 837 {
 838         return afb_evt_event_x2_from_evtid(afb_evt_evtid_create2(prefix, name));
 839 }
 840
 841 /*
 842  * Returns the fullname of the 'eventid'
 843  */
 844 const char *afb_evt_event_x2_fullname(struct afb_event_x2 *eventid)
 845 {
 846         struct afb_evtid *evtid = afb_evt_event_x2_to_evtid(eventid);
 847         return evtid ? evtid->fullname : NULL;
 848 }
 849
 850 /*
 851  * Returns the id of the 'eventid'
 852  */
 853 int afb_evt_event_x2_id(struct afb_event_x2 *eventid)
 854 {
 855         struct afb_evtid *evtid = afb_evt_event_x2_to_evtid(eventid);
 856         return evtid ? evtid->id : 0;
 857 }
 858
 859 /*
 860  * Makes the 'listener' watching 'eventid'
 861  * Returns 0 in case of success or else -1.
 862  */
 863 int afb_evt_event_x2_add_watch(struct afb_evt_listener *listener, struct afb_event_x2 *eventid)
 864 {
 865         struct afb_evtid *evtid = afb_evt_event_x2_to_evtid(eventid);
 866
 867         /* check parameter */
 868         if (!evtid) {
 869                 errno = EINVAL;
 870                 return -1;
 871         }
 872
 873         /* search the existing watch for the listener */
 874         return afb_evt_watch_add_evtid(listener, evtid);
 875 }
 876
 877 /*
 878  * Avoids the 'listener' to watch 'eventid'
 879  * Returns 0 in case of success or else -1.
 880  */
 881 int afb_evt_event_x2_remove_watch(struct afb_evt_listener *listener, struct afb_event_x2 *eventid)
 882 {
 883         struct afb_evtid *evtid = afb_evt_event_x2_to_evtid(eventid);
 884
 885         /* check parameter */
 886         if (!evtid) {
 887                 errno = EINVAL;
 888                 return -1;
 889         }
 890
 891         /* search the existing watch */
 892         return afb_evt_watch_sub_evtid(listener, evtid);
 893 }
 894
 895 int afb_evt_event_x2_push(struct afb_event_x2 *eventid, struct json_object *object)
 896 {
 897         struct afb_evtid *evtid = afb_evt_event_x2_to_evtid(eventid);
 898         if (evtid)
 899                 return afb_evt_evtid_hooked_push(evtid, object);
 900         json_object_put(object);
 901         return 0;
 902 }
 903
 904 int afb_evt_event_x2_unhooked_push(struct afb_event_x2 *eventid, struct json_object *object)
 905 {
 906         struct afb_evtid *evtid = afb_evt_event_x2_to_evtid(eventid);
 907         if (evtid)
 908                 return afb_evt_evtid_push(evtid, object);
 909         json_object_put(object);
 910         return 0;
 911 }
 912
 913 struct afb_event_x1 afb_evt_event_from_evtid(struct afb_evtid *evtid)
 914 {
 915         return evtid
 916                 ? (struct afb_event_x1){ .itf = &afb_evt_hooked_eventid_itf, .closure = &evtid->eventid }
 917                 : (struct afb_event_x1){ .itf = NULL, .closure = NULL };
 918 }
 919
 920 void afb_evt_event_x2_unref(struct afb_event_x2 *eventid)
 921 {
 922         struct afb_evtid *evtid = afb_evt_event_x2_to_evtid(eventid);
 923         if (evtid)
 924                 afb_evt_evtid_unref(evtid);
 925 }
 926
 927 struct afb_event_x2 *afb_evt_event_x2_addref(struct afb_event_x2 *eventid)
 928 {
 929         struct afb_evtid *evtid = afb_evt_event_x2_to_evtid(eventid);
 930         if (evtid)
 931                 afb_evt_evtid_addref(evtid);
 932         return eventid;
 933 }
 934