X-Git-Url: https://gerrit.automotivelinux.org/gerrit/gitweb?a=blobdiff_plain;f=src%2Fafb-stub-ws.c;h=5000f9f98714c97d9463f1368e531c5cb6b55330;hb=9490a0a5fe69f54aa30c4f75a79acde9d06a80f1;hp=44247dd3189723cfd498db1307fcc1d33fc09524;hpb=325e6a7f034c80562096d60ab01f2e4532eea98c;p=src%2Fapp-framework-binder.git diff --git a/src/afb-stub-ws.c b/src/afb-stub-ws.c index 44247dd3..5000f9f9 100644 --- a/src/afb-stub-ws.c +++ b/src/afb-stub-ws.c @@ -1,5 +1,5 @@ /* - * Copyright (C) 2015, 2016, 2017 "IoT.bzh" + * Copyright (C) 2015-2018 "IoT.bzh" * Author José Bollo * * Licensed under the Apache License, Version 2.0 (the "License"); @@ -17,8 +17,6 @@ #define _GNU_SOURCE -#define NO_PLUGIN_VERBOSE_MACRO - #include #include #include @@ -33,11 +31,8 @@ #include #include -#include - -#include -#include "afb-common.h" +#include #include "afb-session.h" #include "afb-cred.h" @@ -49,44 +44,12 @@ #include "afb-evt.h" #include "afb-xreq.h" #include "verbose.h" +#include "fdev.h" #include "jobs.h" struct afb_stub_ws; -/******************* handling subcalls *****************************/ - -/** - * Structure on server side for recording pending - * subcalls. - */ -struct server_subcall -{ - struct server_subcall *next; /**< next subcall for the client */ - uint32_t subcallid; /**< the subcallid */ - void (*callback)(void*, int, struct json_object*); /**< callback on completion */ - void *closure; /**< closure of the callback */ -}; - -/** - * Structure for sending back replies on client side - */ -struct client_subcall -{ - struct afb_stub_ws *stubws; /**< stub descriptor */ - uint32_t subcallid; /**< subcallid for the reply */ -}; - -/* - * structure for recording calls on client side - */ -struct client_call { - struct client_call *next; /* the next call */ - struct afb_stub_ws *stubws; /* the stub_ws */ - struct afb_xreq *xreq; /* the request handle */ - uint32_t msgid; /* the message identifier */ -}; - /* * structure for a ws request */ @@ -102,7 +65,7 @@ struct server_req { struct client_event { struct client_event *next; - struct afb_eventid *eventid; + struct afb_event_x2 *event; int id; int refcount; }; @@ -126,6 +89,15 @@ struct server_describe struct afb_proto_ws_describe *describe; }; +/* + * structure for recording sessions + */ +struct server_session +{ + struct server_session *next; + struct afb_session *session; +}; + /******************* stub description for client or servers ******************/ struct afb_stub_ws @@ -145,9 +117,12 @@ struct afb_stub_ws /* event replica (client side) */ struct client_event *events; - /* credentials (server side) */ + /* credentials of the client (server side) */ struct afb_cred *cred; + /* sessions (server side) */ + struct server_session *sessions; + /* apiset */ struct afb_apiset *apiset; @@ -173,57 +148,37 @@ static void server_req_destroy_cb(struct afb_xreq *xreq) free(wreq); } -static void server_req_success_cb(struct afb_xreq *xreq, struct json_object *obj, const char *info) +static void server_req_reply_cb(struct afb_xreq *xreq, struct json_object *obj, const char *error, const char *info) { int rc; struct server_req *wreq = CONTAINER_OF_XREQ(struct server_req, xreq); - rc = afb_proto_ws_call_success(wreq->call, obj, info); + rc = afb_proto_ws_call_reply(wreq->call, obj, error, info); if (rc < 0) - ERROR("error while sending success"); + ERROR("error while sending reply"); json_object_put(obj); } -static void server_req_fail_cb(struct afb_xreq *xreq, const char *status, const char *info) -{ - int rc; - struct server_req *wreq = CONTAINER_OF_XREQ(struct server_req, xreq); - - rc = afb_proto_ws_call_fail(wreq->call, status, info); - if (rc < 0) - ERROR("error while sending fail"); -} - -static void server_req_subcall_cb(struct afb_xreq *xreq, const char *api, const char *verb, struct json_object *args, void (*callback)(void*, int, struct json_object*), void *cb_closure) -{ - int rc; - struct server_req *wreq = CONTAINER_OF_XREQ(struct server_req, xreq); - - rc = afb_proto_ws_call_subcall(wreq->call, api, verb, args, callback, cb_closure); - if (rc < 0) - ERROR("error while sending subcall"); -} - -static int server_req_subscribe_cb(struct afb_xreq *xreq, struct afb_eventid *event) +static int server_req_subscribe_cb(struct afb_xreq *xreq, struct afb_event_x2 *event) { int rc; struct server_req *wreq = CONTAINER_OF_XREQ(struct server_req, xreq); - rc = afb_evt_add_watch(wreq->stubws->listener, event); + rc = afb_evt_event_x2_add_watch(wreq->stubws->listener, event); if (rc >= 0) - rc = afb_proto_ws_call_subscribe(wreq->call, afb_evt_event_fullname(event), afb_evt_event_id(event)); + rc = afb_proto_ws_call_subscribe(wreq->call, afb_evt_event_x2_fullname(event), afb_evt_event_x2_id(event)); if (rc < 0) ERROR("error while subscribing event"); return rc; } -static int server_req_unsubscribe_cb(struct afb_xreq *xreq, struct afb_eventid *event) +static int server_req_unsubscribe_cb(struct afb_xreq *xreq, struct afb_event_x2 *event) { int rc, rc2; struct server_req *wreq = CONTAINER_OF_XREQ(struct server_req, xreq); - rc = afb_proto_ws_call_unsubscribe(wreq->call, afb_evt_event_fullname(event), afb_evt_event_id(event)); - rc2 = afb_evt_remove_watch(wreq->stubws->listener, event); + rc = afb_proto_ws_call_unsubscribe(wreq->call, afb_evt_event_x2_fullname(event), afb_evt_event_x2_id(event)); + rc2 = afb_evt_event_x2_remove_watch(wreq->stubws->listener, event); if (rc >= 0 && rc2 < 0) rc = rc2; if (rc < 0) @@ -232,10 +187,8 @@ static int server_req_unsubscribe_cb(struct afb_xreq *xreq, struct afb_eventid * } static const struct afb_xreq_query_itf server_req_xreq_itf = { - .success = server_req_success_cb, - .fail = server_req_fail_cb, + .reply = server_req_reply_cb, .unref = server_req_destroy_cb, - .subcall = server_req_subcall_cb, .subscribe = server_req_subscribe_cb, .unsubscribe = server_req_unsubscribe_cb }; @@ -248,7 +201,7 @@ static struct client_event *client_event_search(struct afb_stub_ws *stubws, uint struct client_event *ev; ev = stubws->events; - while (ev != NULL && (ev->id != eventid || 0 != strcmp(afb_evt_event_fullname(ev->eventid), name))) + while (ev != NULL && (ev->id != eventid || 0 != strcmp(afb_evt_event_x2_fullname(ev->event), name))) ev = ev->next; return ev; @@ -257,10 +210,20 @@ static struct client_event *client_event_search(struct afb_stub_ws *stubws, uint /* on call, propagate it to the ws service */ static void client_call_cb(void * closure, struct afb_xreq *xreq) { + int rc; struct afb_stub_ws *stubws = closure; - afb_proto_ws_client_call(stubws->proto, xreq->verb, afb_xreq_json(xreq), afb_session_uuid(xreq->context.session), xreq); - afb_xreq_unhooked_addref(xreq); + rc = afb_proto_ws_client_call( + stubws->proto, + xreq->request.called_verb, + afb_xreq_json(xreq), + afb_session_uuid(xreq->context.session), + xreq, + xreq_on_behalf_cred_export(xreq)); + if (rc >= 0) + afb_xreq_unhooked_addref(xreq); + else + afb_xreq_reply(xreq, NULL, "internal", "can't send message"); } static void client_on_description_cb(void *closure, struct json_object *data) @@ -329,19 +292,11 @@ static void server_event_broadcast(void *closure, const char *event, int eventid /*****************************************************/ -static void on_reply_success(void *closure, void *request, struct json_object *result, const char *info) -{ - struct afb_xreq *xreq = request; - - afb_xreq_success(xreq, result, *info ? info : NULL); - afb_xreq_unhooked_unref(xreq); -} - -static void on_reply_fail(void *closure, void *request, const char *status, const char *info) +static void on_reply(void *closure, void *request, struct json_object *object, const char *error, const char *info) { struct afb_xreq *xreq = request; - afb_xreq_fail(xreq, status, *info ? info : NULL); + afb_xreq_reply(xreq, object, error, info); afb_xreq_unhooked_unref(xreq); } @@ -360,8 +315,8 @@ static void on_event_create(void *closure, const char *event_name, int event_id) /* no conflict, try to add it */ ev = malloc(sizeof *ev); if (ev != NULL) { - ev->eventid = afb_evt_create_event(event_name); - if (ev->eventid != NULL) { + ev->event = afb_evt_event_x2_create(event_name); + if (ev->event != NULL) { ev->refcount = 1; ev->id = event_id; ev->next = stubws->events; @@ -394,7 +349,7 @@ static void on_event_remove(void *closure, const char *event_name, int event_id) *prv = ev->next; /* destroys the event */ - afb_evt_event_unref(ev->eventid); + afb_evt_event_x2_unref(ev->event); free(ev); } @@ -409,7 +364,7 @@ static void on_event_subscribe(void *closure, void *request, const char *event_n if (ev == NULL) return; - if (afb_xreq_subscribe(xreq, ev->eventid) < 0) + if (afb_xreq_subscribe(xreq, ev->event) < 0) ERROR("can't subscribe: %m"); } @@ -424,7 +379,7 @@ static void on_event_unsubscribe(void *closure, void *request, const char *event if (ev == NULL) return; - if (afb_xreq_unsubscribe(xreq, ev->eventid) < 0) + if (afb_xreq_unsubscribe(xreq, ev->event) < 0) ERROR("can't unsubscribe: %m"); } @@ -436,7 +391,7 @@ static void on_event_push(void *closure, const char *event_name, int event_id, s /* check conflicts */ ev = client_event_search(stubws, event_id, event_name); if (ev) - afb_evt_push(ev->eventid, data); + afb_evt_event_x2_push(ev->event, data); else ERROR("unreadable push event"); } @@ -446,22 +401,51 @@ static void on_event_broadcast(void *closure, const char *event_name, struct jso afb_evt_broadcast(event_name, data); } -static void client_subcall_reply_cb(void *closure, int status, json_object *object, struct afb_request *request) +/*****************************************************/ + +static void record_session(struct afb_stub_ws *stubws, struct afb_session *session) { - struct afb_proto_ws_subcall *subcall = closure; - afb_proto_ws_subcall_reply(subcall, status, object); + struct server_session *s, **prv; + + /* search */ + prv = &stubws->sessions; + while ((s = *prv)) { + if (s->session == session) + return; + if (afb_session_is_closed(s->session)) { + *prv = s->next; + afb_session_unref(s->session); + free(s); + } + else + prv = &s->next; + } + + /* create */ + s = malloc(sizeof *s); + if (s) { + s->session = afb_session_addref(session); + s->next = stubws->sessions; + stubws->sessions = s; + } } -static void on_subcall(void *closure, struct afb_proto_ws_subcall *subcall, void *request, const char *api, const char *verb, struct json_object *args) +static void release_all_sessions(struct afb_stub_ws *stubws) { - struct afb_xreq *xreq = request; + struct server_session *s, *n; - afb_xreq_subcall(xreq, api, verb, args, client_subcall_reply_cb, subcall); + s = __atomic_exchange_n(&stubws->sessions, NULL, __ATOMIC_RELAXED); + while(s) { + n = s->next; + afb_session_unref(s->session); + free(s); + s = n; + } } /*****************************************************/ -static void on_call(void *closure, struct afb_proto_ws_call *call, const char *verb, struct json_object *args, const char *sessionid) +static void on_call(void *closure, struct afb_proto_ws_call *call, const char *verb, struct json_object *args, const char *sessionid, const char *user_creds) { struct afb_stub_ws *stubws = closure; struct server_req *wreq; @@ -480,11 +464,15 @@ static void on_call(void *closure, struct afb_proto_ws_call *call, const char *v /* init the context */ if (afb_context_connect(&wreq->xreq.context, sessionid, NULL) < 0) goto unconnected; + wreq->xreq.context.validated = 1; + record_session(stubws, wreq->xreq.context.session); + if (wreq->xreq.context.created) + afb_session_set_autoclose(wreq->xreq.context.session, 1); /* makes the call */ - wreq->xreq.cred = afb_cred_addref(stubws->cred); - wreq->xreq.api = stubws->apiname; - wreq->xreq.verb = verb; + wreq->xreq.cred = afb_cred_mixed_on_behalf_import(stubws->cred, sessionid, user_creds); + wreq->xreq.request.called_api = stubws->apiname; + wreq->xreq.request.called_verb = verb; wreq->xreq.json = args; afb_xreq_process(&wreq->xreq, stubws->apiset); return; @@ -494,7 +482,7 @@ unconnected: out_of_memory: json_object_put(args); afb_stub_ws_unref(stubws); - afb_proto_ws_call_fail(call, "internal-error", NULL); + afb_proto_ws_call_reply(call, NULL, "internal-error", NULL); afb_proto_ws_call_unref(call); } @@ -544,15 +532,13 @@ static void on_describe(void *closure, struct afb_proto_ws_describe *describe) static const struct afb_proto_ws_client_itf client_itf = { - .on_reply_success = on_reply_success, - .on_reply_fail = on_reply_fail, + .on_reply = on_reply, .on_event_create = on_event_create, .on_event_remove = on_event_remove, .on_event_subscribe = on_event_subscribe, .on_event_unsubscribe = on_event_unsubscribe, .on_event_push = on_event_push, .on_event_broadcast = on_event_broadcast, - .on_subcall = on_subcall }; static const struct afb_proto_ws_server_itf server_itf = @@ -585,7 +571,7 @@ static void drop_all_events(struct afb_stub_ws *stubws) while (ev) { nxt = ev->next; - afb_evt_event_unref(ev->eventid); + afb_evt_event_x2_unref(ev->event); free(ev); ev = nxt; } @@ -596,13 +582,22 @@ static void on_hangup(void *closure) { struct afb_stub_ws *stubws = closure; + afb_stub_ws_addref(stubws); if (stubws->on_hangup) stubws->on_hangup(stubws); + + release_all_sessions(stubws); + afb_stub_ws_unref(stubws); +} + +static int enqueue_processing(void (*callback)(int signum, void* arg), void *arg) +{ + return jobs_queue(NULL, 0, callback, arg); } /*****************************************************/ -static struct afb_stub_ws *afb_stub_ws_create(int fd, const char *apiname, struct afb_apiset *apiset, int client) +static struct afb_stub_ws *afb_stub_ws_create(struct fdev *fdev, const char *apiname, struct afb_apiset *apiset, int client) { struct afb_stub_ws *stubws; @@ -611,33 +606,36 @@ static struct afb_stub_ws *afb_stub_ws_create(int fd, const char *apiname, struc errno = ENOMEM; else { if (client) - stubws->proto = afb_proto_ws_create_client(fd, &client_itf, stubws); + stubws->proto = afb_proto_ws_create_client(fdev, &client_itf, stubws); else - stubws->proto = afb_proto_ws_create_server(fd, &server_itf, stubws); - if (stubws->proto != NULL) { + stubws->proto = afb_proto_ws_create_server(fdev, &server_itf, stubws); + + if (stubws->proto) { strcpy(stubws->apiname, apiname); stubws->apiset = afb_apiset_addref(apiset); stubws->refcount = 1; afb_proto_ws_on_hangup(stubws->proto, on_hangup); + afb_proto_ws_set_queuing(stubws->proto, enqueue_processing); return stubws; } free(stubws); } + fdev_unref(fdev); return NULL; } -struct afb_stub_ws *afb_stub_ws_create_client(int fd, const char *apiname, struct afb_apiset *apiset) +struct afb_stub_ws *afb_stub_ws_create_client(struct fdev *fdev, const char *apiname, struct afb_apiset *apiset) { - return afb_stub_ws_create(fd, apiname, apiset, 1); + return afb_stub_ws_create(fdev, apiname, apiset, 1); } -struct afb_stub_ws *afb_stub_ws_create_server(int fd, const char *apiname, struct afb_apiset *apiset) +struct afb_stub_ws *afb_stub_ws_create_server(struct fdev *fdev, const char *apiname, struct afb_apiset *apiset) { struct afb_stub_ws *stubws; - stubws = afb_stub_ws_create(fd, apiname, apiset, 0); + stubws = afb_stub_ws_create(fdev, apiname, apiset, 0); if (stubws) { - stubws->cred = afb_cred_create_for_socket(fd); + stubws->cred = afb_cred_create_for_socket(fdev_fd(fdev)); stubws->listener = afb_evt_listener_create(&server_evt_itf, stubws); if (stubws->listener != NULL) return stubws; @@ -650,7 +648,9 @@ void afb_stub_ws_unref(struct afb_stub_ws *stubws) { if (!__atomic_sub_fetch(&stubws->refcount, 1, __ATOMIC_RELAXED)) { drop_all_events(stubws); - afb_evt_listener_unref(stubws->listener); + if (stubws->listener) + afb_evt_listener_unref(stubws->listener); + release_all_sessions(stubws); afb_proto_ws_unref(stubws->proto); afb_cred_unref(stubws->cred); afb_apiset_unref(stubws->apiset); @@ -663,7 +663,7 @@ void afb_stub_ws_addref(struct afb_stub_ws *stubws) __atomic_add_fetch(&stubws->refcount, 1, __ATOMIC_RELAXED); } -void afb_stub_ws_on_hangup(struct afb_stub_ws *stubws, void (*on_hangup)(struct afb_stub_ws*)) +void afb_stub_ws_set_on_hangup(struct afb_stub_ws *stubws, void (*on_hangup)(struct afb_stub_ws*)) { stubws->on_hangup = on_hangup; } @@ -673,9 +673,9 @@ const char *afb_stub_ws_name(struct afb_stub_ws *stubws) return stubws->apiname; } -struct afb_api afb_stub_ws_client_api(struct afb_stub_ws *stubws) +struct afb_api_item afb_stub_ws_client_api(struct afb_stub_ws *stubws) { - struct afb_api api; + struct afb_api_item api; assert(!stubws->listener); /* check client */ api.closure = stubws;