websocket: Tune maximum received length
[src/app-framework-binder.git] / 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  * Returns the count of listener taht received the event.
183  */
184 static int evt_push(struct afb_evt_event *evt, struct json_object *obj)
185 {
186         int result;
187         struct afb_evt_watch *watch;
188         struct afb_evt_listener *listener;
189
190         result = 0;
191         pthread_mutex_lock(&evt->mutex);
192         if (evt->hookflags & afb_hook_flag_evt_push_before)
193                 afb_hook_evt_push_before(evt->name, evt->id, obj);
194         watch = evt->watchs;
195         while(watch) {
196                 listener = watch->listener;
197                 assert(listener->itf->push != NULL);
198                 if (watch->activity != 0) {
199                         listener->itf->push(listener->closure, evt->name, evt->id, json_object_get(obj));
200                         result++;
201                 }
202                 watch = watch->next_by_event;
203         }
204         if (evt->hookflags & afb_hook_flag_evt_push_after)
205                 afb_hook_evt_push_after(evt->name, evt->id, obj, result);
206         pthread_mutex_unlock(&evt->mutex);
207         json_object_put(obj);
208         return result;
209 }
210
211 /*
212  * Returns the name associated to the event 'evt'.
213  */
214 static const char *evt_name(struct afb_evt_event *evt)
215 {
216         const char *name = strchr(evt->name, '/');
217         name = name ? name + 1 : evt->name;
218         if (evt->hookflags & afb_hook_flag_evt_name)
219                 afb_hook_evt_name(evt->name, evt->id);
220         return name;
221 }
222
223 /*
224  * remove the 'watch'
225  */
226 static void remove_watch(struct afb_evt_watch *watch)
227 {
228         struct afb_evt_watch **prv;
229         struct afb_evt_event *evt;
230         struct afb_evt_listener *listener;
231
232         /* notify listener if needed */
233         evt = watch->event;
234         listener = watch->listener;
235         if (watch->activity != 0 && listener->itf->remove != NULL)
236                 listener->itf->remove(listener->closure, evt->name, evt->id);
237
238         /* unlink the watch for its event */
239         prv = &evt->watchs;
240         while(*prv != watch)
241                 prv = &(*prv)->next_by_event;
242         *prv = watch->next_by_event;
243
244         /* unlink the watch for its listener */
245         prv = &listener->watchs;
246         while(*prv != watch)
247                 prv = &(*prv)->next_by_listener;
248         *prv = watch->next_by_listener;
249
250         /* recycle memory */
251         free(watch);
252 }
253
254 /*
255  * Destroys the event 'evt'
256  */
257 static void evt_destroy(struct afb_evt_event *evt)
258 {
259         int found;
260         struct afb_evt_event **prv;
261         struct afb_evt_listener *listener;
262
263         if (evt != NULL) {
264                 /* unlinks the event if valid! */
265                 pthread_mutex_lock(&events_mutex);
266                 found = 0;
267                 prv = &events;
268                 while (*prv && !(found = (*prv == evt)))
269                         prv = &(*prv)->next;
270                 if (found)
271                         *prv = evt->next;
272                 pthread_mutex_unlock(&events_mutex);
273
274                 /* destroys the event */
275                 if (found) {
276                         /* removes all watchers */
277                         while(evt->watchs != NULL) {
278                                 listener = evt->watchs->listener;
279                                 pthread_mutex_lock(&listener->mutex);
280                                 pthread_mutex_lock(&evt->mutex);
281                                 remove_watch(evt->watchs);
282                                 pthread_mutex_unlock(&evt->mutex);
283                                 pthread_mutex_unlock(&listener->mutex);
284                         }
285
286                         /* hook */
287                         if (evt->hookflags & afb_hook_flag_evt_drop)
288                                 afb_hook_evt_drop(evt->name, evt->id);
289
290                         /* free */
291                         pthread_mutex_destroy(&evt->mutex);
292                         free(evt);
293                 }
294         }
295 }
296
297 /*
298  * Creates an event of 'name' and returns it.
299  * Returns an event with closure==NULL in case of error.
300  */
301 struct afb_event afb_evt_create_event(const char *name)
302 {
303         size_t len;
304         struct afb_evt_event *evt, *oevt;
305
306         /* allocates the event */
307         len = strlen(name);
308         evt = malloc(len + sizeof * evt);
309         if (evt == NULL)
310                 goto error;
311
312         /* allocates the id */
313         pthread_mutex_lock(&events_mutex);
314         do {
315                 if (++event_id_counter < 0) {
316                         event_id_wrapped = 1;
317                         event_id_counter = 1024; /* heuristic: small numbers are not destroyed */
318                 }
319                 if (!event_id_wrapped)
320                         break;
321                 oevt = events;
322                 while(oevt != NULL && oevt->id != event_id_counter)
323                         oevt = oevt->next;
324         } while (oevt != NULL);
325
326         /* initialize the event */
327         memcpy(evt->name, name, len + 1);
328         evt->next = events;
329         evt->watchs = NULL;
330         evt->id = event_id_counter;
331         pthread_mutex_init(&evt->mutex, NULL);
332         events = evt;
333         evt->hookflags = afb_hook_flags_evt(evt->name);
334         if (evt->hookflags & afb_hook_flag_evt_create)
335                 afb_hook_evt_create(evt->name, evt->id);
336         pthread_mutex_unlock(&events_mutex);
337
338         /* returns the event */
339         return (struct afb_event){ .itf = &afb_evt_event_itf, .closure = evt };
340 error:
341         return (struct afb_event){ .itf = NULL, .closure = NULL };
342 }
343
344 /*
345  * Returns the name of the 'event'
346  */
347 const char *afb_evt_event_name(struct afb_event event)
348 {
349         return (event.itf != &afb_evt_event_itf) ? NULL : ((struct afb_evt_event *)event.closure)->name;
350 }
351
352 /*
353  * Returns the id of the 'event'
354  */
355 int afb_evt_event_id(struct afb_event event)
356 {
357         return (event.itf != &afb_evt_event_itf) ? 0 : ((struct afb_evt_event *)event.closure)->id;
358 }
359
360 /*
361  * Returns an instance of the listener defined by the 'send' callback
362  * and the 'closure'.
363  * Returns NULL in case of memory depletion.
364  */
365 struct afb_evt_listener *afb_evt_listener_create(const struct afb_evt_itf *itf, void *closure)
366 {
367         struct afb_evt_listener *listener;
368
369         /* search if an instance already exists */
370         pthread_mutex_lock(&listeners_mutex);
371         listener = listeners;
372         while (listener != NULL) {
373                 if (listener->itf == itf && listener->closure == closure) {
374                         listener = afb_evt_listener_addref(listener);
375                         goto found;
376                 }
377                 listener = listener->next;
378         }
379
380         /* allocates */
381         listener = calloc(1, sizeof *listener);
382         if (listener != NULL) {
383                 /* init */
384                 listener->itf = itf;
385                 listener->closure = closure;
386                 listener->watchs = NULL;
387                 listener->refcount = 1;
388                 pthread_mutex_init(&listener->mutex, NULL);
389                 listener->next = listeners;
390                 listeners = listener;
391         }
392  found:
393         pthread_mutex_unlock(&listeners_mutex);
394         return listener;
395 }
396
397 /*
398  * Increases the reference count of 'listener' and returns it
399  */
400 struct afb_evt_listener *afb_evt_listener_addref(struct afb_evt_listener *listener)
401 {
402         __atomic_add_fetch(&listener->refcount, 1, __ATOMIC_RELAXED);
403         return listener;
404 }
405
406 /*
407  * Decreases the reference count of the 'listener' and destroys it
408  * when no more used.
409  */
410 void afb_evt_listener_unref(struct afb_evt_listener *listener)
411 {
412         struct afb_evt_listener **prv;
413         struct afb_evt_event *evt;
414
415         if (!__atomic_sub_fetch(&listener->refcount, 1, __ATOMIC_RELAXED)) {
416
417                 /* unlink the listener */
418                 pthread_mutex_lock(&listeners_mutex);
419                 prv = &listeners;
420                 while (*prv != listener)
421                         prv = &(*prv)->next;
422                 *prv = listener->next;
423                 pthread_mutex_unlock(&listeners_mutex);
424
425                 /* remove the watchers */
426                 pthread_mutex_lock(&listener->mutex);
427                 while (listener->watchs != NULL) {
428                         evt = listener->watchs->event;
429                         pthread_mutex_lock(&evt->mutex);
430                         remove_watch(listener->watchs);
431                         pthread_mutex_unlock(&evt->mutex);
432                 }
433                 pthread_mutex_unlock(&listener->mutex);
434
435                 /* free the listener */
436                 pthread_mutex_destroy(&listener->mutex);
437                 free(listener);
438         }
439 }
440
441 /*
442  * Makes the 'listener' watching 'event'
443  * Returns 0 in case of success or else -1.
444  */
445 int afb_evt_add_watch(struct afb_evt_listener *listener, struct afb_event event)
446 {
447         struct afb_evt_watch *watch;
448         struct afb_evt_event *evt;
449
450         /* check parameter */
451         if (event.itf != &afb_evt_event_itf || listener->itf->push == NULL) {
452                 errno = EINVAL;
453                 return -1;
454         }
455
456         /* search the existing watch for the listener */
457         evt = event.closure;
458         pthread_mutex_lock(&listener->mutex);
459         watch = listener->watchs;
460         while(watch != NULL) {
461                 if (watch->event == evt)
462                         goto found;
463                 watch = watch->next_by_listener;
464         }
465
466         /* not found, allocate a new */
467         watch = malloc(sizeof *watch);
468         if (watch == NULL) {
469                 pthread_mutex_unlock(&listener->mutex);
470                 errno = ENOMEM;
471                 return -1;
472         }
473
474         /* initialise and link */
475         watch->event = evt;
476         watch->activity = 0;
477         watch->listener = listener;
478         watch->next_by_listener = listener->watchs;
479         listener->watchs = watch;
480         pthread_mutex_lock(&evt->mutex);
481         watch->next_by_event = evt->watchs;
482         evt->watchs = watch;
483         pthread_mutex_unlock(&evt->mutex);
484
485 found:
486         if (watch->activity == 0 && listener->itf->add != NULL)
487                 listener->itf->add(listener->closure, evt->name, evt->id);
488         watch->activity++;
489         pthread_mutex_unlock(&listener->mutex);
490
491         return 0;
492 }
493
494 /*
495  * Avoids the 'listener' to watch 'event'
496  * Returns 0 in case of success or else -1.
497  */
498 int afb_evt_remove_watch(struct afb_evt_listener *listener, struct afb_event event)
499 {
500         struct afb_evt_watch *watch;
501         struct afb_evt_event *evt;
502
503         /* check parameter */
504         if (event.itf != &afb_evt_event_itf) {
505                 errno = EINVAL;
506                 return -1;
507         }
508
509         /* search the existing watch */
510         evt = event.closure;
511         pthread_mutex_lock(&listener->mutex);
512         watch = listener->watchs;
513         while(watch != NULL) {
514                 if (watch->event == evt) {
515                         if (watch->activity != 0) {
516                                 watch->activity--;
517                                 if (watch->activity == 0 && listener->itf->remove != NULL)
518                                         listener->itf->remove(listener->closure, evt->name, evt->id);
519                         }
520                         pthread_mutex_unlock(&listener->mutex);
521                         return 0;
522                 }
523                 watch = watch->next_by_listener;
524         }
525         pthread_mutex_unlock(&listener->mutex);
526         errno = ENOENT;
527         return -1;
528 }
529
530 /*
531  * update the hooks for events
532  */
533 void afb_evt_update_hooks()
534 {
535         struct afb_evt_event *evt;
536
537         pthread_mutex_lock(&events_mutex);
538         for (evt = events ; evt ; evt = evt->next)
539                 evt->hookflags = afb_hook_flags_evt(evt->name);
540         pthread_mutex_unlock(&events_mutex);
541 }
542