afb-stub-ws: Enforce asynchronous describe
[src/app-framework-binder.git] / src / afb-stub-ws.c
1 /*
2  * Copyright (C) 2015-2019 "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 <fcntl.h>
24 #include <unistd.h>
25 #include <errno.h>
26 #include <endian.h>
27 #include <netdb.h>
28 #include <sys/types.h>
29 #include <sys/socket.h>
30 #include <sys/un.h>
31 #include <pthread.h>
32
33 #include <json-c/json.h>
34
35 #include <afb/afb-event-x2.h>
36
37 #include "afb-session.h"
38 #include "afb-cred.h"
39 #include "afb-api.h"
40 #include "afb-apiset.h"
41 #include "afb-proto-ws.h"
42 #include "afb-stub-ws.h"
43 #include "afb-context.h"
44 #include "afb-evt.h"
45 #include "afb-xreq.h"
46 #include "verbose.h"
47 #include "fdev.h"
48 #include "jobs.h"
49
50 struct afb_stub_ws;
51
52
53 /**
54  * structure for a ws request: requests on server side
55  */
56 struct server_req {
57         struct afb_xreq xreq;           /**< the xreq */
58         struct afb_stub_ws *stubws;     /**< the client of the request */
59         struct afb_proto_ws_call *call; /**< the incoming call */
60 };
61
62 /**
63  * structure for recording events on client side
64  */
65 struct client_event
66 {
67         struct client_event *next;      /**< link to the next */
68         struct afb_event_x2 *event;     /**< the local event */
69         int id;                         /**< the identifier */
70         int refcount;                   /**< a reference count */
71 };
72
73 /**
74  * structure for jobs of describing
75  */
76 struct server_describe
77 {
78         struct afb_stub_ws *stubws;
79         struct afb_proto_ws_describe *describe;
80 };
81
82 /**
83  * structure for recording sessions
84  */
85 struct server_session
86 {
87         struct server_session *next;
88         struct afb_session *session;
89 };
90
91 /******************* stub description for client or servers ******************/
92
93 struct afb_stub_ws
94 {
95         /* protocol */
96         struct afb_proto_ws *proto;
97
98         /* apiset */
99         struct afb_apiset *apiset;
100
101         /* on hangup callback */
102         void (*on_hangup)(struct afb_stub_ws *);
103
104         union {
105                 /* server side */
106                 struct {
107                         /* listener for events */
108                         struct afb_evt_listener *listener;
109
110                         /* sessions */
111                         struct server_session *sessions;
112
113                         /* credentials of the client */
114                         struct afb_cred *cred;
115                 };
116
117                 /* client side */
118                 struct {
119                         /* event replica  */
120                         struct client_event *events;
121
122                         /* robustify */
123                         struct {
124                                 struct fdev *(*reopen)(void*);
125                                 void *closure;
126                                 void (*release)(void*);
127                         } robust;
128                 };
129         };
130
131         /* count of references */
132         unsigned refcount;
133
134         /* type of the stub: 0=server, 1=client */
135         uint8_t is_client;
136
137         /* the api name */
138         char apiname[];
139 };
140
141 static struct afb_proto_ws *afb_stub_ws_create_proto(struct afb_stub_ws *stubws, struct fdev *fdev, uint8_t server);
142
143 /******************* ws request part for server *****************/
144
145 /* decrement the reference count of the request and free/release it on falling to null */
146 static void server_req_destroy_cb(struct afb_xreq *xreq)
147 {
148         struct server_req *wreq = CONTAINER_OF_XREQ(struct server_req, xreq);
149
150         afb_context_disconnect(&wreq->xreq.context);
151         afb_cred_unref(wreq->xreq.cred);
152         json_object_put(wreq->xreq.json);
153         afb_proto_ws_call_unref(wreq->call);
154         afb_stub_ws_unref(wreq->stubws);
155         free(wreq);
156 }
157
158 static void server_req_reply_cb(struct afb_xreq *xreq, struct json_object *obj, const char *error, const char *info)
159 {
160         int rc;
161         struct server_req *wreq = CONTAINER_OF_XREQ(struct server_req, xreq);
162
163         rc = afb_proto_ws_call_reply(wreq->call, obj, error, info);
164         if (rc < 0)
165                 ERROR("error while sending reply");
166         json_object_put(obj);
167 }
168
169 static int server_req_subscribe_cb(struct afb_xreq *xreq, struct afb_event_x2 *event)
170 {
171         int rc;
172         struct server_req *wreq = CONTAINER_OF_XREQ(struct server_req, xreq);
173
174         rc = afb_evt_event_x2_add_watch(wreq->stubws->listener, event);
175         if (rc >= 0)
176                 rc = afb_proto_ws_call_subscribe(wreq->call,  afb_evt_event_x2_fullname(event), afb_evt_event_x2_id(event));
177         if (rc < 0)
178                 ERROR("error while subscribing event");
179         return rc;
180 }
181
182 static int server_req_unsubscribe_cb(struct afb_xreq *xreq, struct afb_event_x2 *event)
183 {
184         int rc, rc2;
185         struct server_req *wreq = CONTAINER_OF_XREQ(struct server_req, xreq);
186
187         rc = afb_proto_ws_call_unsubscribe(wreq->call,  afb_evt_event_x2_fullname(event), afb_evt_event_x2_id(event));
188         rc2 = afb_evt_event_x2_remove_watch(wreq->stubws->listener, event);
189         if (rc >= 0 && rc2 < 0)
190                 rc = rc2;
191         if (rc < 0)
192                 ERROR("error while unsubscribing event");
193         return rc;
194 }
195
196 static const struct afb_xreq_query_itf server_req_xreq_itf = {
197         .reply = server_req_reply_cb,
198         .unref = server_req_destroy_cb,
199         .subscribe = server_req_subscribe_cb,
200         .unsubscribe = server_req_unsubscribe_cb
201 };
202
203 /******************* client part **********************************/
204
205 /* destroy all events */
206 static void client_drop_all_events(struct afb_stub_ws *stubws)
207 {
208         struct client_event *ev, *nxt;
209
210         nxt = __atomic_exchange_n(&stubws->events, NULL, __ATOMIC_RELAXED);
211         while (nxt) {
212                 ev = nxt;
213                 nxt = ev->next;
214                 afb_evt_event_x2_unref(ev->event);
215                 free(ev);
216         }
217 }
218
219 /* search the event */
220 static struct client_event *client_event_search(struct afb_stub_ws *stubws, uint32_t eventid, const char *name)
221 {
222         struct client_event *ev;
223
224         ev = stubws->events;
225         while (ev != NULL && (ev->id != eventid || 0 != strcmp(afb_evt_event_x2_fullname(ev->event), name)))
226                 ev = ev->next;
227
228         DEBUG("searching event %s[%d]: %s", name, eventid, ev ? "found" : "not found");
229         return ev;
230 }
231
232 static struct afb_proto_ws *client_get_proto(struct afb_stub_ws *stubws)
233 {
234         struct fdev *fdev;
235         struct afb_proto_ws *proto;
236
237         proto = stubws->proto;
238         if (proto == NULL && stubws->robust.reopen) {
239                 fdev = stubws->robust.reopen(stubws->robust.closure);
240                 if (fdev != NULL)
241                         proto = afb_stub_ws_create_proto(stubws, fdev, 0);
242         }
243         return proto;
244 }
245
246 /* on call, propagate it to the ws service */
247 static void client_api_call_cb(void * closure, struct afb_xreq *xreq)
248 {
249         int rc;
250         struct afb_stub_ws *stubws = closure;
251         struct afb_proto_ws *proto;
252
253         proto = client_get_proto(stubws);
254         if (proto == NULL) {
255                 afb_xreq_reply(xreq, NULL, "disconnected", "server hung up");
256                 return;
257         }
258
259         afb_xreq_unhooked_addref(xreq);
260         rc = afb_proto_ws_client_call(
261                         proto,
262                         xreq->request.called_verb,
263                         afb_xreq_json(xreq),
264                         afb_session_uuid(xreq->context.session),
265                         xreq,
266                         xreq_on_behalf_cred_export(xreq));
267         if (rc < 0) {
268                 afb_xreq_reply(xreq, NULL, "internal", "can't send message");
269                 afb_xreq_unhooked_unref(xreq);
270         }
271 }
272
273 /* get the description */
274 static void client_api_describe_cb(void * closure, void (*describecb)(void *, struct json_object *), void *clocb)
275 {
276         struct afb_stub_ws *stubws = closure;
277         struct afb_proto_ws *proto;
278
279         proto = client_get_proto(stubws);
280         if (proto)
281                 afb_proto_ws_client_describe(proto, describecb, clocb);
282         else
283                 describecb(clocb, NULL);
284 }
285
286 /******************* server part: manage events **********************************/
287
288 static void server_event_add_cb(void *closure, const char *event, uint16_t eventid)
289 {
290         struct afb_stub_ws *stubws = closure;
291
292         if (stubws->proto != NULL)
293                 afb_proto_ws_server_event_create(stubws->proto, event, eventid);
294 }
295
296 static void server_event_remove_cb(void *closure, const char *event, uint16_t eventid)
297 {
298         struct afb_stub_ws *stubws = closure;
299
300         if (stubws->proto != NULL)
301                 afb_proto_ws_server_event_remove(stubws->proto, event, eventid);
302 }
303
304 static void server_event_push_cb(void *closure, const char *event, uint16_t eventid, struct json_object *object)
305 {
306         struct afb_stub_ws *stubws = closure;
307
308         if (stubws->proto != NULL)
309                 afb_proto_ws_server_event_push(stubws->proto, event, eventid, object);
310         json_object_put(object);
311 }
312
313 static void server_event_broadcast_cb(void *closure, const char *event, struct json_object *object, const uuid_binary_t uuid, uint8_t hop)
314 {
315         struct afb_stub_ws *stubws = closure;
316
317         if (stubws->proto != NULL)
318                 afb_proto_ws_server_event_broadcast(stubws->proto, event, object, uuid, hop);
319         json_object_put(object);
320 }
321
322 /*****************************************************/
323
324 static void client_on_reply_cb(void *closure, void *request, struct json_object *object, const char *error, const char *info)
325 {
326         struct afb_xreq *xreq = request;
327
328         afb_xreq_reply(xreq, object, error, info);
329         afb_xreq_unhooked_unref(xreq);
330 }
331
332 static void client_on_event_create_cb(void *closure, const char *event_name, int event_id)
333 {
334         struct afb_stub_ws *stubws = closure;
335         struct client_event *ev;
336
337         /* check conflicts */
338         ev = client_event_search(stubws, event_id, event_name);
339         if (ev != NULL) {
340                 __atomic_add_fetch(&ev->refcount, 1, __ATOMIC_RELAXED);
341                 return;
342         }
343
344         /* no conflict, try to add it */
345         ev = malloc(sizeof *ev);
346         if (ev != NULL) {
347                 ev->event = afb_evt_event_x2_create(event_name);
348                 if (ev->event != NULL) {
349                         ev->refcount = 1;
350                         ev->id = event_id;
351                         ev->next = stubws->events;
352                         stubws->events = ev;
353                         return;
354                 }
355                 free(ev);
356         }
357         ERROR("can't create event %s, out of memory", event_name);
358 }
359
360 static void client_on_event_remove_cb(void *closure, const char *event_name, int event_id)
361 {
362         struct afb_stub_ws *stubws = closure;
363         struct client_event *ev, **prv;
364
365         /* check conflicts */
366         ev = client_event_search(stubws, event_id, event_name);
367         if (ev == NULL)
368                 return;
369
370         /* decrease the reference count */
371
372         if (__atomic_sub_fetch(&ev->refcount, 1, __ATOMIC_RELAXED))
373                 return;
374
375         /* unlinks the event */
376         prv = &stubws->events;
377         while (*prv != ev)
378                 prv = &(*prv)->next;
379         *prv = ev->next;
380
381         /* destroys the event */
382         afb_evt_event_x2_unref(ev->event);
383         free(ev);
384 }
385
386 static void client_on_event_subscribe_cb(void *closure, void *request, const char *event_name, int event_id)
387 {
388         struct afb_stub_ws *stubws = closure;
389         struct afb_xreq *xreq = request;
390         struct client_event *ev;
391
392         /* check conflicts */
393         ev = client_event_search(stubws, event_id, event_name);
394         if (ev == NULL)
395                 return;
396
397         if (afb_xreq_subscribe(xreq, ev->event) < 0)
398                 ERROR("can't subscribe: %m");
399 }
400
401 static void client_on_event_unsubscribe_cb(void *closure, void *request, const char *event_name, int event_id)
402 {
403         struct afb_stub_ws *stubws = closure;
404         struct afb_xreq *xreq = request;
405         struct client_event *ev;
406
407         /* check conflicts */
408         ev = client_event_search(stubws, event_id, event_name);
409         if (ev == NULL)
410                 return;
411
412         if (afb_xreq_unsubscribe(xreq, ev->event) < 0)
413                 ERROR("can't unsubscribe: %m");
414 }
415
416 static void client_on_event_push_cb(void *closure, const char *event_name, int event_id, struct json_object *data)
417 {
418         struct afb_stub_ws *stubws = closure;
419         struct client_event *ev;
420
421         /* check conflicts */
422         ev = client_event_search(stubws, event_id, event_name);
423         if (ev)
424                 afb_evt_event_x2_push(ev->event, data);
425         else
426                 ERROR("unreadable push event");
427 }
428
429 static void client_on_event_broadcast_cb(void *closure, const char *event_name, struct json_object *data, const uuid_binary_t uuid, uint8_t hop)
430 {
431         afb_evt_rebroadcast(event_name, data, uuid, hop);
432 }
433
434 /*****************************************************/
435
436 static void server_record_session(struct afb_stub_ws *stubws, struct afb_session *session)
437 {
438         struct server_session *s, **prv;
439
440         /* search */
441         prv = &stubws->sessions;
442         while ((s = *prv)) {
443                 if (s->session == session)
444                         return;
445                 if (afb_session_is_closed(s->session)) {
446                         *prv = s->next;
447                         afb_session_unref(s->session);
448                         free(s);
449                 }
450                 else
451                         prv = &s->next;
452         }
453
454         /* create */
455         s = malloc(sizeof *s);
456         if (s) {
457                 s->session = afb_session_addref(session);
458                 s->next = stubws->sessions;
459                 stubws->sessions = s;
460         }
461 }
462
463 static void server_release_all_sessions(struct afb_stub_ws *stubws)
464 {
465         struct server_session *ses, *nses;
466
467         nses = __atomic_exchange_n(&stubws->sessions, NULL, __ATOMIC_RELAXED);
468         while(nses) {
469                 ses = nses;
470                 nses = ses->next;
471                 afb_session_unref(ses->session);
472                 free(ses);
473         }
474 }
475
476 /*****************************************************/
477
478 static void server_on_call_cb(void *closure, struct afb_proto_ws_call *call, const char *verb, struct json_object *args, const char *sessionid, const char *user_creds)
479 {
480         struct afb_stub_ws *stubws = closure;
481         struct server_req *wreq;
482
483         afb_stub_ws_addref(stubws);
484
485         /* create the request */
486         wreq = malloc(sizeof *wreq);
487         if (wreq == NULL)
488                 goto out_of_memory;
489
490         afb_xreq_init(&wreq->xreq, &server_req_xreq_itf);
491         wreq->stubws = stubws;
492         wreq->call = call;
493
494         /* init the context */
495         if (afb_context_connect_validated(&wreq->xreq.context, sessionid) < 0)
496                 goto unconnected;
497         server_record_session(stubws, wreq->xreq.context.session);
498         if (wreq->xreq.context.created)
499                 afb_session_set_autoclose(wreq->xreq.context.session, 1);
500
501         /* makes the call */
502         wreq->xreq.cred = afb_cred_mixed_on_behalf_import(stubws->cred, &wreq->xreq.context, user_creds);
503         wreq->xreq.request.called_api = stubws->apiname;
504         wreq->xreq.request.called_verb = verb;
505         wreq->xreq.json = args;
506         afb_xreq_process(&wreq->xreq, stubws->apiset);
507         return;
508
509 unconnected:
510         free(wreq);
511 out_of_memory:
512         json_object_put(args);
513         afb_stub_ws_unref(stubws);
514         afb_proto_ws_call_reply(call, NULL, "internal-error", NULL);
515         afb_proto_ws_call_unref(call);
516 }
517
518 static void server_on_description_cb(void *closure, struct json_object *description)
519 {
520         struct afb_proto_ws_describe *describe = closure;
521         afb_proto_ws_describe_put(describe, description);
522         json_object_put(description);
523 }
524
525
526 static void server_on_describe_cb(void *closure, struct afb_proto_ws_describe *describe)
527 {
528         struct afb_stub_ws *stubws = closure;
529
530         afb_apiset_describe(stubws->apiset, stubws->apiname, server_on_description_cb, describe);
531 }
532
533 /*****************************************************/
534
535 static const struct afb_proto_ws_client_itf client_itf =
536 {
537         .on_reply = client_on_reply_cb,
538         .on_event_create = client_on_event_create_cb,
539         .on_event_remove = client_on_event_remove_cb,
540         .on_event_subscribe = client_on_event_subscribe_cb,
541         .on_event_unsubscribe = client_on_event_unsubscribe_cb,
542         .on_event_push = client_on_event_push_cb,
543         .on_event_broadcast = client_on_event_broadcast_cb,
544 };
545
546 static struct afb_api_itf client_api_itf = {
547         .call = client_api_call_cb,
548         .describe = client_api_describe_cb
549 };
550
551 static const struct afb_proto_ws_server_itf server_itf =
552 {
553         .on_call = server_on_call_cb,
554         .on_describe = server_on_describe_cb
555 };
556
557 /* the interface for events pushing */
558 static const struct afb_evt_itf server_event_itf = {
559         .broadcast = server_event_broadcast_cb,
560         .push = server_event_push_cb,
561         .add = server_event_add_cb,
562         .remove = server_event_remove_cb
563 };
564
565 /*****************************************************/
566
567 /* disconnect */
568 static void disconnect(struct afb_stub_ws *stubws)
569 {
570         afb_proto_ws_unref(__atomic_exchange_n(&stubws->proto, NULL, __ATOMIC_RELAXED));
571         if (stubws->is_client)
572                 client_drop_all_events(stubws);
573         else {
574                 afb_evt_listener_unref(__atomic_exchange_n(&stubws->listener, NULL, __ATOMIC_RELAXED));
575                 afb_cred_unref(__atomic_exchange_n(&stubws->cred, NULL, __ATOMIC_RELAXED));
576                 server_release_all_sessions(stubws);
577         }
578 }
579
580
581 /* callback when receiving a hangup */
582 static void on_hangup(void *closure)
583 {
584         struct afb_stub_ws *stubws = closure;
585
586         if (stubws->proto) {
587                 afb_stub_ws_addref(stubws);
588                 disconnect(stubws);
589                 if (stubws->on_hangup)
590                         stubws->on_hangup(stubws);
591                 afb_stub_ws_unref(stubws);
592         }
593 }
594
595 static int enqueue_processing(struct afb_proto_ws *proto, void (*callback)(int signum, void* arg), void *arg)
596 {
597         return jobs_queue(proto, 0, callback, arg);
598 }
599
600 /*****************************************************/
601
602 static struct afb_proto_ws *afb_stub_ws_create_proto(struct afb_stub_ws *stubws, struct fdev *fdev, uint8_t is_client)
603 {
604         struct afb_proto_ws *proto;
605
606         stubws->proto = proto = is_client
607                   ? afb_proto_ws_create_client(fdev, &client_itf, stubws)
608                   : afb_proto_ws_create_server(fdev, &server_itf, stubws);
609         if (proto) {
610                 afb_proto_ws_on_hangup(proto, on_hangup);
611                 afb_proto_ws_set_queuing(proto, enqueue_processing);
612         }
613
614         return proto;
615 }
616
617 static struct afb_stub_ws *afb_stub_ws_create(struct fdev *fdev, const char *apiname, struct afb_apiset *apiset, uint8_t is_client)
618 {
619         struct afb_stub_ws *stubws;
620
621         stubws = calloc(1, sizeof *stubws + 1 + strlen(apiname));
622         if (stubws == NULL)
623                 errno = ENOMEM;
624         else {
625                 if (afb_stub_ws_create_proto(stubws, fdev, is_client)) {
626                         stubws->refcount = 1;
627                         stubws->is_client = is_client;
628                         strcpy(stubws->apiname, apiname);
629                         stubws->apiset = afb_apiset_addref(apiset);
630                         return stubws;
631                 }
632                 free(stubws);
633         }
634         fdev_unref(fdev);
635         return NULL;
636 }
637
638 struct afb_stub_ws *afb_stub_ws_create_client(struct fdev *fdev, const char *apiname, struct afb_apiset *apiset)
639 {
640         return afb_stub_ws_create(fdev, apiname, apiset, 1);
641 }
642
643 struct afb_stub_ws *afb_stub_ws_create_server(struct fdev *fdev, const char *apiname, struct afb_apiset *apiset)
644 {
645         struct afb_stub_ws *stubws;
646
647         stubws = afb_stub_ws_create(fdev, apiname, apiset, 0);
648         if (stubws) {
649                 stubws->cred = afb_cred_create_for_socket(fdev_fd(fdev));
650                 stubws->listener = afb_evt_listener_create(&server_event_itf, stubws);
651                 if (stubws->listener != NULL)
652                         return stubws;
653                 afb_stub_ws_unref(stubws);
654         }
655         return NULL;
656 }
657
658 void afb_stub_ws_unref(struct afb_stub_ws *stubws)
659 {
660         if (stubws && !__atomic_sub_fetch(&stubws->refcount, 1, __ATOMIC_RELAXED)) {
661
662                 if (stubws->is_client) {
663                         stubws->robust.reopen = NULL;
664                         if (stubws->robust.release)
665                                 stubws->robust.release(stubws->robust.closure);
666                 }
667
668                 disconnect(stubws);
669                 afb_apiset_unref(stubws->apiset);
670                 free(stubws);
671         }
672 }
673
674 void afb_stub_ws_addref(struct afb_stub_ws *stubws)
675 {
676         __atomic_add_fetch(&stubws->refcount, 1, __ATOMIC_RELAXED);
677 }
678
679 void afb_stub_ws_set_on_hangup(struct afb_stub_ws *stubws, void (*on_hangup)(struct afb_stub_ws*))
680 {
681         stubws->on_hangup = on_hangup;
682 }
683
684 const char *afb_stub_ws_name(struct afb_stub_ws *stubws)
685 {
686         return stubws->apiname;
687 }
688
689 struct afb_api_item afb_stub_ws_client_api(struct afb_stub_ws *stubws)
690 {
691         struct afb_api_item api;
692
693         assert(stubws->is_client); /* check client */
694         api.closure = stubws;
695         api.itf = &client_api_itf;
696         api.group = stubws; /* serialize for reconnections */
697         return api;
698 }
699
700 int afb_stub_ws_client_add(struct afb_stub_ws *stubws, struct afb_apiset *apiset)
701 {
702         return afb_apiset_add(apiset, stubws->apiname, afb_stub_ws_client_api(stubws));
703 }
704
705 void afb_stub_ws_client_robustify(struct afb_stub_ws *stubws, struct fdev *(*reopen)(void*), void *closure, void (*release)(void*))
706 {
707         assert(stubws->is_client); /* check client */
708
709         if (stubws->robust.release)
710                 stubws->robust.release(stubws->robust.closure);
711
712         stubws->robust.reopen = reopen;
713         stubws->robust.closure = closure;
714         stubws->robust.release = release;
715 }