summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
-rw-r--r--src/afb-evt.c234
-rw-r--r--src/afb-evt.h1
-rw-r--r--src/afb-stub-ws.c5
3 files changed, 134 insertions, 106 deletions
diff --git a/src/afb-evt.c b/src/afb-evt.c
index 400f8fbe..7122a33d 100644
--- a/src/afb-evt.c
+++ b/src/afb-evt.c
@@ -87,9 +87,6 @@ struct afb_evtid {
/* id of the event */
uint16_t id;
- /* has client? */
- int has_client;
-
/* fullname of the event */
char fullname[];
};
@@ -110,9 +107,6 @@ struct afb_evt_watch {
/* link to the next watcher for the same listener */
struct afb_evt_watch *next_by_listener;
-
- /* activity */
- unsigned activity;
};
/*
@@ -412,23 +406,17 @@ int afb_evt_broadcast(const char *event, struct json_object *object)
*/
static void push_evtid(struct afb_evtid *evtid, struct json_object *object)
{
- int has_client;
struct afb_evt_watch *watch;
struct afb_evt_listener *listener;
- has_client = 0;
pthread_rwlock_rdlock(&evtid->rwlock);
watch = evtid->watchs;
while(watch) {
listener = watch->listener;
assert(listener->itf->push != NULL);
- if (watch->activity != 0) {
- listener->itf->push(listener->closure, evtid->fullname, evtid->id, json_object_get(object));
- has_client = 1;
- }
+ listener->itf->push(listener->closure, evtid->fullname, evtid->id, json_object_get(object));
watch = watch->next_by_evtid;
}
- evtid->has_client = has_client;
pthread_rwlock_unlock(&evtid->rwlock);
}
@@ -455,6 +443,9 @@ int afb_evt_evtid_push(struct afb_evtid *evtid, struct json_object *object)
struct job_evtid *je;
int rc;
+ if (!evtid->watchs)
+ return 0;
+
je = make_job_evtid(evtid, object);
if (je == NULL) {
ERROR("Cant't create push evtid job item for %s(%s)",
@@ -465,7 +456,7 @@ int afb_evt_evtid_push(struct afb_evtid *evtid, struct json_object *object)
rc = jobs_queue(PUSH_JOB_GROUP, 0, push_job_evtid, je);
if (rc == 0)
- rc = evtid->has_client;
+ rc = 1;
else {
ERROR("cant't queue push evtid job item for %s(%s)",
evtid->fullname, json_object_to_json_string(object));
@@ -507,32 +498,54 @@ int afb_evt_evtid_hooked_push(struct afb_evtid *evtid, struct json_object *obj)
}
#endif
-/*
- * remove the 'watch'
- */
-static void remove_watch(struct afb_evt_watch *watch)
+static void unwatch(struct afb_evt_listener *listener, struct afb_evtid *evtid, int remove)
+{
+ /* notify listener if needed */
+ if (remove && listener->itf->remove != NULL)
+ listener->itf->remove(listener->closure, evtid->fullname, evtid->id);
+}
+
+static void evtid_unwatch(struct afb_evtid *evtid, struct afb_evt_listener *listener, struct afb_evt_watch *watch, int remove)
{
struct afb_evt_watch **prv;
- struct afb_evtid *evtid;
- struct afb_evt_listener *listener;
/* notify listener if needed */
- evtid = watch->evtid;
- listener = watch->listener;
- if (watch->activity != 0 && listener->itf->remove != NULL)
- listener->itf->remove(listener->closure, evtid->fullname, evtid->id);
+ unwatch(listener, evtid, remove);
/* unlink the watch for its event */
- prv = &evtid->watchs;
- while(*prv != watch)
- prv = &(*prv)->next_by_evtid;
- *prv = watch->next_by_evtid;
-
- /* unlink the watch for its listener */
+ pthread_rwlock_wrlock(&listener->rwlock);
prv = &listener->watchs;
- while(*prv != watch)
+ while(*prv) {
+ if (*prv == watch) {
+ *prv = watch->next_by_listener;
+ break;
+ }
prv = &(*prv)->next_by_listener;
- *prv = watch->next_by_listener;
+ }
+ pthread_rwlock_unlock(&listener->rwlock);
+
+ /* recycle memory */
+ free(watch);
+}
+
+static void listener_unwatch(struct afb_evt_listener *listener, struct afb_evtid *evtid, struct afb_evt_watch *watch, int remove)
+{
+ struct afb_evt_watch **prv;
+
+ /* notify listener if needed */
+ unwatch(listener, evtid, remove);
+
+ /* unlink the watch for its event */
+ pthread_rwlock_wrlock(&evtid->rwlock);
+ prv = &evtid->watchs;
+ while(*prv) {
+ if (*prv == watch) {
+ *prv = watch->next_by_evtid;
+ break;
+ }
+ prv = &(*prv)->next_by_evtid;
+ }
+ pthread_rwlock_unlock(&evtid->rwlock);
/* recycle memory */
free(watch);
@@ -578,7 +591,6 @@ struct afb_evtid *afb_evt_evtid_create(const char *fullname)
evtid->refcount = 1;
evtid->watchs = NULL;
evtid->id = id;
- evtid->has_client = 0;
pthread_rwlock_init(&evtid->rwlock, NULL);
evtids = evtid;
#if WITH_AFB_HOOK
@@ -644,41 +656,42 @@ struct afb_evtid *afb_evt_evtid_hooked_addref(struct afb_evtid *evtid)
*/
void afb_evt_evtid_unref(struct afb_evtid *evtid)
{
- int found;
- struct afb_evtid **prv;
- struct afb_evt_listener *listener;
+ struct afb_evtid **prv, *oev;
+ struct afb_evt_watch *watch, *nwatch;
if (!__atomic_sub_fetch(&evtid->refcount, 1, __ATOMIC_RELAXED)) {
/* unlinks the event if valid! */
pthread_rwlock_wrlock(&events_rwlock);
- found = 0;
prv = &evtids;
- while (*prv && !(found = (*prv == evtid)))
- prv = &(*prv)->next;
- if (found) {
- *prv = evtid->next;
- event_count--;
+ for(;;) {
+ oev = *prv;
+ if (oev == evtid)
+ break;
+ if (!oev) {
+ ERROR("unexpected event");
+ pthread_rwlock_unlock(&events_rwlock);
+ return;
+ }
+ prv = &oev->next;
}
+ event_count--;
+ *prv = evtid->next;
pthread_rwlock_unlock(&events_rwlock);
- /* destroys the event */
- if (!found)
- ERROR("event not found");
- else {
- /* removes all watchers */
- while(evtid->watchs != NULL) {
- listener = evtid->watchs->listener;
- pthread_rwlock_wrlock(&listener->rwlock);
- pthread_rwlock_wrlock(&evtid->rwlock);
- remove_watch(evtid->watchs);
- pthread_rwlock_unlock(&evtid->rwlock);
- pthread_rwlock_unlock(&listener->rwlock);
- }
-
- /* free */
- pthread_rwlock_destroy(&evtid->rwlock);
- free(evtid);
+ /* removes all watchers */
+ pthread_rwlock_wrlock(&evtid->rwlock);
+ watch = evtid->watchs;
+ evtid->watchs = NULL;
+ pthread_rwlock_unlock(&evtid->rwlock);
+ while(watch) {
+ nwatch = watch->next_by_evtid;
+ evtid_unwatch(evtid, watch->listener, watch, 1);
+ watch = nwatch;
}
+
+ /* free */
+ pthread_rwlock_destroy(&evtid->rwlock);
+ free(evtid);
}
}
@@ -785,28 +798,29 @@ struct afb_evt_listener *afb_evt_listener_addref(struct afb_evt_listener *listen
*/
void afb_evt_listener_unref(struct afb_evt_listener *listener)
{
- struct afb_evt_listener **prv;
- struct afb_evtid *evtid;
+ struct afb_evt_listener **prv, *olis;
if (listener && !__atomic_sub_fetch(&listener->refcount, 1, __ATOMIC_RELAXED)) {
/* unlink the listener */
pthread_rwlock_wrlock(&listeners_rwlock);
prv = &listeners;
- while (*prv != listener)
- prv = &(*prv)->next;
+ for(;;) {
+ olis = *prv;
+ if (olis == listener)
+ break;
+ if (!olis) {
+ ERROR("unexpected listener");
+ pthread_rwlock_unlock(&listeners_rwlock);
+ return;
+ }
+ prv = &olis->next;
+ }
*prv = listener->next;
pthread_rwlock_unlock(&listeners_rwlock);
/* remove the watchers */
- pthread_rwlock_wrlock(&listener->rwlock);
- while (listener->watchs != NULL) {
- evtid = listener->watchs->evtid;
- pthread_rwlock_wrlock(&evtid->rwlock);
- remove_watch(listener->watchs);
- pthread_rwlock_unlock(&evtid->rwlock);
- }
- pthread_rwlock_unlock(&listener->rwlock);
+ afb_evt_listener_unwatch_all(listener, 0);
/* free the listener */
pthread_rwlock_destroy(&listener->rwlock);
@@ -833,7 +847,7 @@ int afb_evt_listener_watch_evt(struct afb_evt_listener *listener, struct afb_evt
watch = listener->watchs;
while(watch != NULL) {
if (watch->evtid == evtid)
- goto found;
+ goto end;
watch = watch->next_by_listener;
}
@@ -847,7 +861,6 @@ int afb_evt_listener_watch_evt(struct afb_evt_listener *listener, struct afb_evt
/* initialise and link */
watch->evtid = evtid;
- watch->activity = 0;
watch->listener = listener;
watch->next_by_listener = listener->watchs;
listener->watchs = watch;
@@ -856,13 +869,10 @@ int afb_evt_listener_watch_evt(struct afb_evt_listener *listener, struct afb_evt
evtid->watchs = watch;
pthread_rwlock_unlock(&evtid->rwlock);
-found:
- if (watch->activity == 0 && listener->itf->add != NULL)
+ if (listener->itf->add != NULL)
listener->itf->add(listener->closure, evtid->fullname, evtid->id);
- watch->activity++;
- evtid->has_client = 1;
+end:
pthread_rwlock_unlock(&listener->rwlock);
-
return 0;
}
@@ -872,26 +882,26 @@ found:
*/
int afb_evt_listener_unwatch_evt(struct afb_evt_listener *listener, struct afb_evtid *evtid)
{
- struct afb_evt_watch *watch;
+ struct afb_evt_watch *watch, **pwatch;
/* search the existing watch */
pthread_rwlock_wrlock(&listener->rwlock);
- watch = listener->watchs;
- while(watch != NULL) {
- if (watch->evtid == evtid) {
- if (watch->activity != 0) {
- watch->activity--;
- if (watch->activity == 0 && listener->itf->remove != NULL)
- listener->itf->remove(listener->closure, evtid->fullname, evtid->id);
- }
+ pwatch = &listener->watchs;
+ for (;;) {
+ watch = *pwatch;
+ if (!watch) {
+ pthread_rwlock_unlock(&listener->rwlock);
+ errno = ENOENT;
+ return -1;
+ }
+ if (evtid == watch->evtid) {
+ *pwatch = watch->next_by_listener;
pthread_rwlock_unlock(&listener->rwlock);
+ listener_unwatch(listener, evtid, watch, 1);
return 0;
}
- watch = watch->next_by_listener;
+ pwatch = &watch->next_by_listener;
}
- pthread_rwlock_unlock(&listener->rwlock);
- errno = ENOENT;
- return -1;
}
/*
@@ -900,28 +910,48 @@ int afb_evt_listener_unwatch_evt(struct afb_evt_listener *listener, struct afb_e
*/
int afb_evt_listener_unwatch_id(struct afb_evt_listener *listener, uint16_t eventid)
{
- struct afb_evt_watch *watch;
+ struct afb_evt_watch *watch, **pwatch;
struct afb_evtid *evtid;
/* search the existing watch */
pthread_rwlock_wrlock(&listener->rwlock);
- watch = listener->watchs;
- while(watch != NULL) {
+ pwatch = &listener->watchs;
+ for (;;) {
+ watch = *pwatch;
+ if (!watch) {
+ pthread_rwlock_unlock(&listener->rwlock);
+ errno = ENOENT;
+ return -1;
+ }
evtid = watch->evtid;
if (evtid->id == eventid) {
- if (watch->activity != 0) {
- watch->activity--;
- if (watch->activity == 0 && listener->itf->remove != NULL)
- listener->itf->remove(listener->closure, evtid->fullname, evtid->id);
- }
+ *pwatch = watch->next_by_listener;
pthread_rwlock_unlock(&listener->rwlock);
+ listener_unwatch(listener, evtid, watch, 1);
return 0;
}
- watch = watch->next_by_listener;
+ pwatch = &watch->next_by_listener;
}
+}
+
+/*
+ * Avoids the 'listener' to watch any event, calling the callback
+ * 'remove' of the interface if 'remoe' is not zero.
+ */
+void afb_evt_listener_unwatch_all(struct afb_evt_listener *listener, int remove)
+{
+ struct afb_evt_watch *watch, *nwatch;
+
+ /* search the existing watch */
+ pthread_rwlock_wrlock(&listener->rwlock);
+ watch = listener->watchs;
+ listener->watchs = NULL;
pthread_rwlock_unlock(&listener->rwlock);
- errno = ENOENT;
- return -1;
+ while(watch) {
+ nwatch = watch->next_by_listener;
+ listener_unwatch(listener, watch->evtid, watch, remove);
+ watch = nwatch;
+ }
}
#if WITH_AFB_HOOK
diff --git a/src/afb-evt.h b/src/afb-evt.h
index 88308aaa..e4c54eed 100644
--- a/src/afb-evt.h
+++ b/src/afb-evt.h
@@ -61,6 +61,7 @@ extern int afb_evt_evtid_broadcast(struct afb_evtid *evtid, struct json_object *
extern int afb_evt_listener_watch_evt(struct afb_evt_listener *listener, struct afb_evtid *evtid);
extern int afb_evt_listener_unwatch_evt(struct afb_evt_listener *listener, struct afb_evtid *evtid);
extern int afb_evt_listener_unwatch_id(struct afb_evt_listener *listener, uint16_t eventid);
+extern void afb_evt_listener_unwatch_all(struct afb_evt_listener *listener, int remove);
extern struct afb_event_x2 *afb_evt_event_x2_create(const char *fullname);
extern struct afb_event_x2 *afb_evt_event_x2_create2(const char *prefix, const char *name);
diff --git a/src/afb-stub-ws.c b/src/afb-stub-ws.c
index 197d4cdc..806c5e47 100644
--- a/src/afb-stub-ws.c
+++ b/src/afb-stub-ws.c
@@ -177,13 +177,10 @@ static int server_req_subscribe_cb(struct afb_xreq *xreq, struct afb_event_x2 *e
static int server_req_unsubscribe_cb(struct afb_xreq *xreq, struct afb_event_x2 *event)
{
- int rc, rc2;
+ int rc;
struct server_req *wreq = CONTAINER_OF_XREQ(struct server_req, xreq);
rc = afb_proto_ws_call_unsubscribe(wreq->call, afb_evt_event_x2_id(event));
- rc2 = afb_evt_listener_unwatch_x2(wreq->stubws->listener, event);
- if (rc >= 0 && rc2 < 0)
- rc = rc2;
if (rc < 0)
ERROR("error while unsubscribing event");
return rc;