X-Git-Url: https://gerrit.automotivelinux.org/gerrit/gitweb?a=blobdiff_plain;f=src%2Fafb-stub-ws.c;h=2a48745bb013579a552cd44c0c9bb42387402fa5;hb=eaf5670e9d10b5d7c066043e7563706cf1e01bd5;hp=7637f22c003656cb721d0f7ba257041a2779cc44;hpb=7682c2aacb3efd6abed3dee43f8a03d7646d8153;p=src%2Fapp-framework-binder.git diff --git a/src/afb-stub-ws.c b/src/afb-stub-ws.c index 7637f22c..2a48745b 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-2019 "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,75 +44,43 @@ #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 + * structure for a ws request: requests on server side */ struct server_req { - struct afb_xreq xreq; /* the xreq */ - struct afb_stub_ws *stubws; /* the client of the request */ - struct afb_proto_ws_call *call; /* the incoming call */ + struct afb_xreq xreq; /**< the xreq */ + struct afb_stub_ws *stubws; /**< the client of the request */ + struct afb_proto_ws_call *call; /**< the incoming call */ }; -/* +/** * structure for recording events on client side */ struct client_event { - struct client_event *next; - struct afb_eventid *eventid; - int id; - int refcount; + struct client_event *next; /**< link to the next */ + struct afb_event_x2 *event; /**< the local event */ + int id; /**< the identifier */ + int refcount; /**< a reference count */ }; -/* - * structure for recording describe requests +/** + * structure for recording describe requests on the client side */ struct client_describe { - struct afb_stub_ws *stubws; - struct jobloop *jobloop; - struct json_object *result; + struct afb_stub_ws *stubws; /**< the stub */ + struct jobloop *jobloop; /**< the jobloop to leave */ + struct json_object *result; /**< result */ }; -/* +/** * structure for jobs of describing */ struct server_describe @@ -126,38 +89,67 @@ 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 { - /* count of references */ - int refcount; - - /* resource control */ - pthread_mutex_t mutex; - /* protocol */ struct afb_proto_ws *proto; - /* listener for events (server side) */ - struct afb_evt_listener *listener; - - /* event replica (client side) */ - struct client_event *events; - - /* credentials (server side) */ - struct afb_cred *cred; - /* apiset */ struct afb_apiset *apiset; /* on hangup callback */ void (*on_hangup)(struct afb_stub_ws *); + union { + /* server side */ + struct { + /* listener for events */ + struct afb_evt_listener *listener; + + /* sessions */ + struct server_session *sessions; + + /* credentials of the client */ + struct afb_cred *cred; + }; + + /* client side */ + struct { + /* event replica */ + struct client_event *events; + + /* robustify */ + struct { + struct fdev *(*reopen)(void*); + void *closure; + void (*release)(void*); + } robust; + }; + }; + + /* count of references */ + unsigned refcount; + + /* type of the stub: 0=server, 1=client */ + uint8_t is_client; + /* the api name */ - char apiname[1]; + char apiname[]; }; +static struct afb_proto_ws *afb_stub_ws_create_proto(struct afb_stub_ws *stubws, struct fdev *fdev, uint8_t server); + /******************* ws request part for server *****************/ /* decrement the reference count of the request and free/release it on falling to null */ @@ -173,57 +165,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) +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_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) -{ - int rc; - struct server_req *wreq = CONTAINER_OF_XREQ(struct server_req, xreq); - - rc = afb_evt_eventid_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_eventid_fullname(event), afb_evt_eventid_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_eventid_fullname(event), afb_evt_eventid_id(event)); - rc2 = afb_evt_eventid_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,35 +204,80 @@ 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 }; /******************* client part **********************************/ +/* destroy all events */ +static void client_drop_all_events(struct afb_stub_ws *stubws) +{ + struct client_event *ev, *nxt; + + nxt = __atomic_exchange_n(&stubws->events, NULL, __ATOMIC_RELAXED); + while (nxt) { + ev = nxt; + nxt = ev->next; + afb_evt_event_x2_unref(ev->event); + free(ev); + } +} + /* search the event */ static struct client_event *client_event_search(struct afb_stub_ws *stubws, uint32_t eventid, const char *name) { struct client_event *ev; ev = stubws->events; - while (ev != NULL && (ev->id != eventid || 0 != strcmp(afb_evt_eventid_fullname(ev->eventid), name))) + while (ev != NULL && (ev->id != eventid || 0 != strcmp(afb_evt_event_x2_fullname(ev->event), name))) ev = ev->next; + DEBUG("searching event %s[%d]: %s", name, eventid, ev ? "found" : "not found"); return ev; } +static struct afb_proto_ws *client_get_proto(struct afb_stub_ws *stubws) +{ + struct fdev *fdev; + struct afb_proto_ws *proto; + + proto = stubws->proto; + if (proto == NULL && stubws->robust.reopen) { + fdev = stubws->robust.reopen(stubws->robust.closure); + if (fdev != NULL) + proto = afb_stub_ws_create_proto(stubws, fdev, 0); + } + return proto; +} + /* on call, propagate it to the ws service */ -static void client_call_cb(void * closure, struct afb_xreq *xreq) +static void client_api_call_cb(void * closure, struct afb_xreq *xreq) { + int rc; struct afb_stub_ws *stubws = closure; + struct afb_proto_ws *proto; + + proto = client_get_proto(stubws); + if (proto == NULL) { + afb_xreq_reply(xreq, NULL, "disconnected", "server hung up"); + return; + } - afb_proto_ws_client_call(stubws->proto, xreq->request.verb, afb_xreq_json(xreq), afb_session_uuid(xreq->context.session), xreq); afb_xreq_unhooked_addref(xreq); + rc = afb_proto_ws_client_call( + 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_reply(xreq, NULL, "internal", "can't send message"); + afb_xreq_unhooked_unref(xreq); + } } static void client_on_description_cb(void *closure, struct json_object *data) @@ -274,17 +291,19 @@ static void client_on_description_cb(void *closure, struct json_object *data) static void client_send_describe_cb(int signum, void *closure, struct jobloop *jobloop) { struct client_describe *desc = closure; + struct afb_proto_ws *proto; - if (signum) + proto = client_get_proto(desc->stubws); + if (signum || proto == NULL) jobs_leave(jobloop); else { desc->jobloop = jobloop; - afb_proto_ws_client_describe(desc->stubws->proto, client_on_description_cb, desc); + afb_proto_ws_client_describe(proto, client_on_description_cb, desc); } } /* get the description */ -static struct json_object *client_describe_cb(void * closure) +static struct json_object *client_api_describe_cb(void * closure) { struct client_describe desc; @@ -297,55 +316,51 @@ static struct json_object *client_describe_cb(void * closure) /******************* server part: manage events **********************************/ -static void server_event_add(void *closure, const char *event, int eventid) +static void server_event_add_cb(void *closure, const char *event, uint16_t eventid) { struct afb_stub_ws *stubws = closure; - afb_proto_ws_server_event_create(stubws->proto, event, eventid); + if (stubws->proto != NULL) + afb_proto_ws_server_event_create(stubws->proto, event, eventid); } -static void server_event_remove(void *closure, const char *event, int eventid) +static void server_event_remove_cb(void *closure, const char *event, uint16_t eventid) { struct afb_stub_ws *stubws = closure; - afb_proto_ws_server_event_remove(stubws->proto, event, eventid); + if (stubws->proto != NULL) + afb_proto_ws_server_event_remove(stubws->proto, event, eventid); } -static void server_event_push(void *closure, const char *event, int eventid, struct json_object *object) +static void server_event_push_cb(void *closure, const char *event, uint16_t eventid, struct json_object *object) { struct afb_stub_ws *stubws = closure; - afb_proto_ws_server_event_push(stubws->proto, event, eventid, object); + if (stubws->proto != NULL) + afb_proto_ws_server_event_push(stubws->proto, event, eventid, object); json_object_put(object); } -static void server_event_broadcast(void *closure, const char *event, int eventid, struct json_object *object) +static void server_event_broadcast_cb(void *closure, const char *event, struct json_object *object, const uuid_binary_t uuid, uint8_t hop) { struct afb_stub_ws *stubws = closure; - afb_proto_ws_server_event_broadcast(stubws->proto, event, object); + if (stubws->proto != NULL) + afb_proto_ws_server_event_broadcast(stubws->proto, event, object, uuid, hop); json_object_put(object); } /*****************************************************/ -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 client_on_reply_cb(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); } -static void on_event_create(void *closure, const char *event_name, int event_id) +static void client_on_event_create_cb(void *closure, const char *event_name, int event_id) { struct afb_stub_ws *stubws = closure; struct client_event *ev; @@ -353,15 +368,15 @@ static void on_event_create(void *closure, const char *event_name, int event_id) /* check conflicts */ ev = client_event_search(stubws, event_id, event_name); if (ev != NULL) { - ev->refcount++; + __atomic_add_fetch(&ev->refcount, 1, __ATOMIC_RELAXED); return; } /* no conflict, try to add it */ ev = malloc(sizeof *ev); if (ev != NULL) { - ev->eventid = afb_evt_eventid_create(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; @@ -373,7 +388,7 @@ static void on_event_create(void *closure, const char *event_name, int event_id) ERROR("can't create event %s, out of memory", event_name); } -static void on_event_remove(void *closure, const char *event_name, int event_id) +static void client_on_event_remove_cb(void *closure, const char *event_name, int event_id) { struct afb_stub_ws *stubws = closure; struct client_event *ev, **prv; @@ -384,7 +399,8 @@ static void on_event_remove(void *closure, const char *event_name, int event_id) return; /* decrease the reference count */ - if (--ev->refcount) + + if (__atomic_sub_fetch(&ev->refcount, 1, __ATOMIC_RELAXED)) return; /* unlinks the event */ @@ -394,11 +410,11 @@ static void on_event_remove(void *closure, const char *event_name, int event_id) *prv = ev->next; /* destroys the event */ - afb_evt_eventid_unref(ev->eventid); + afb_evt_event_x2_unref(ev->event); free(ev); } -static void on_event_subscribe(void *closure, void *request, const char *event_name, int event_id) +static void client_on_event_subscribe_cb(void *closure, void *request, const char *event_name, int event_id) { struct afb_stub_ws *stubws = closure; struct afb_xreq *xreq = request; @@ -409,11 +425,11 @@ 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"); } -static void on_event_unsubscribe(void *closure, void *request, const char *event_name, int event_id) +static void client_on_event_unsubscribe_cb(void *closure, void *request, const char *event_name, int event_id) { struct afb_stub_ws *stubws = closure; struct afb_xreq *xreq = request; @@ -424,11 +440,11 @@ 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"); } -static void on_event_push(void *closure, const char *event_name, int event_id, struct json_object *data) +static void client_on_event_push_cb(void *closure, const char *event_name, int event_id, struct json_object *data) { struct afb_stub_ws *stubws = closure; struct client_event *ev; @@ -436,32 +452,61 @@ 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_eventid_push(ev->eventid, data); + afb_evt_event_x2_push(ev->event, data); else ERROR("unreadable push event"); } -static void on_event_broadcast(void *closure, const char *event_name, struct json_object *data) +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) { - afb_evt_broadcast(event_name, data); + afb_evt_rebroadcast(event_name, data, uuid, hop); } -static void client_subcall_reply_cb(void *closure, int status, json_object *object, struct afb_request *request) +/*****************************************************/ + +static void server_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 server_release_all_sessions(struct afb_stub_ws *stubws) { - struct afb_xreq *xreq = request; + struct server_session *ses, *nses; - afb_xreq_subcall(xreq, api, verb, args, client_subcall_reply_cb, subcall); + nses = __atomic_exchange_n(&stubws->sessions, NULL, __ATOMIC_RELAXED); + while(nses) { + ses = nses; + nses = ses->next; + afb_session_unref(ses->session); + free(ses); + } } /*****************************************************/ -static void on_call(void *closure, struct afb_proto_ws_call *call, const char *verb, struct json_object *args, const char *sessionid) +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) { struct afb_stub_ws *stubws = closure; struct server_req *wreq; @@ -478,13 +523,16 @@ static void on_call(void *closure, struct afb_proto_ws_call *call, const char *v wreq->call = call; /* init the context */ - if (afb_context_connect(&wreq->xreq.context, sessionid, NULL) < 0) + if (afb_context_connect_validated(&wreq->xreq.context, sessionid) < 0) goto unconnected; + server_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.request.api = stubws->apiname; - wreq->xreq.request.verb = verb; + wreq->xreq.cred = afb_cred_mixed_on_behalf_import(stubws->cred, &wreq->xreq.context, 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,11 +542,11 @@ 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); } -static void server_describe_sjob(int signum, void *closure) +static void server_describe_cb(int signum, void *closure) { struct json_object *obj; struct server_describe *desc = closure; @@ -514,11 +562,11 @@ static void server_describe_sjob(int signum, void *closure) static void server_describe_job(int signum, void *closure) { - server_describe_sjob(signum, closure); + server_describe_cb(signum, closure); free(closure); } -static void on_describe(void *closure, struct afb_proto_ws_describe *describe) +static void server_on_describe_cb(void *closure, struct afb_proto_ws_describe *describe) { struct server_describe *desc, sdesc; struct afb_stub_ws *stubws = closure; @@ -532,113 +580,128 @@ static void on_describe(void *closure, struct afb_proto_ws_describe *describe) afb_stub_ws_addref(stubws); /* process */ - if (desc == &sdesc) - jobs_call(NULL, 0, server_describe_sjob, desc); - else { - if (jobs_queue(NULL, 0, server_describe_job, desc) < 0) - jobs_call(NULL, 0, server_describe_job, desc); - } + if (desc == &sdesc || jobs_queue(NULL, 0, server_describe_job, desc) < 0) + jobs_call(NULL, 0, server_describe_cb, desc); } /*****************************************************/ static const struct afb_proto_ws_client_itf client_itf = { - .on_reply_success = on_reply_success, - .on_reply_fail = on_reply_fail, - .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 + .on_reply = client_on_reply_cb, + .on_event_create = client_on_event_create_cb, + .on_event_remove = client_on_event_remove_cb, + .on_event_subscribe = client_on_event_subscribe_cb, + .on_event_unsubscribe = client_on_event_unsubscribe_cb, + .on_event_push = client_on_event_push_cb, + .on_event_broadcast = client_on_event_broadcast_cb, }; -static const struct afb_proto_ws_server_itf server_itf = -{ - .on_call = on_call, - .on_describe = on_describe +static struct afb_api_itf client_api_itf = { + .call = client_api_call_cb, + .describe = client_api_describe_cb }; -static struct afb_api_itf ws_api_itf = { - .call = client_call_cb, - .describe = client_describe_cb +static const struct afb_proto_ws_server_itf server_itf = +{ + .on_call = server_on_call_cb, + .on_describe = server_on_describe_cb }; /* the interface for events pushing */ -static const struct afb_evt_itf server_evt_itf = { - .broadcast = server_event_broadcast, - .push = server_event_push, - .add = server_event_add, - .remove = server_event_remove +static const struct afb_evt_itf server_event_itf = { + .broadcast = server_event_broadcast_cb, + .push = server_event_push_cb, + .add = server_event_add_cb, + .remove = server_event_remove_cb }; /*****************************************************/ -static void drop_all_events(struct afb_stub_ws *stubws) +/* disconnect */ +static void disconnect(struct afb_stub_ws *stubws) { - struct client_event *ev, *nxt; - - ev = stubws->events; - stubws->events = NULL; - - while (ev) { - nxt = ev->next; - afb_evt_eventid_unref(ev->eventid); - free(ev); - ev = nxt; + afb_proto_ws_unref(__atomic_exchange_n(&stubws->proto, NULL, __ATOMIC_RELAXED)); + if (stubws->is_client) + client_drop_all_events(stubws); + else { + afb_evt_listener_unref(__atomic_exchange_n(&stubws->listener, NULL, __ATOMIC_RELAXED)); + afb_cred_unref(__atomic_exchange_n(&stubws->cred, NULL, __ATOMIC_RELAXED)); + server_release_all_sessions(stubws); } } + /* callback when receiving a hangup */ static void on_hangup(void *closure) { struct afb_stub_ws *stubws = closure; - if (stubws->on_hangup) - stubws->on_hangup(stubws); + if (stubws->proto) { + afb_stub_ws_addref(stubws); + disconnect(stubws); + if (stubws->on_hangup) + stubws->on_hangup(stubws); + afb_stub_ws_unref(stubws); + } +} + +static int enqueue_processing(struct afb_proto_ws *proto, void (*callback)(int signum, void* arg), void *arg) +{ + return jobs_queue(proto, 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_proto_ws *afb_stub_ws_create_proto(struct afb_stub_ws *stubws, struct fdev *fdev, uint8_t is_client) +{ + struct afb_proto_ws *proto; + + stubws->proto = proto = is_client + ? afb_proto_ws_create_client(fdev, &client_itf, stubws) + : afb_proto_ws_create_server(fdev, &server_itf, stubws); + if (proto) { + afb_proto_ws_on_hangup(proto, on_hangup); + afb_proto_ws_set_queuing(proto, enqueue_processing); + } + + return proto; +} + +static struct afb_stub_ws *afb_stub_ws_create(struct fdev *fdev, const char *apiname, struct afb_apiset *apiset, uint8_t is_client) { struct afb_stub_ws *stubws; - stubws = calloc(1, sizeof *stubws + strlen(apiname)); + stubws = calloc(1, sizeof *stubws + 1 + strlen(apiname)); if (stubws == NULL) errno = ENOMEM; else { - if (client) - stubws->proto = afb_proto_ws_create_client(fd, &client_itf, stubws); - else - stubws->proto = afb_proto_ws_create_server(fd, &server_itf, stubws); - if (stubws->proto != NULL) { + if (afb_stub_ws_create_proto(stubws, fdev, is_client)) { + stubws->refcount = 1; + stubws->is_client = is_client; strcpy(stubws->apiname, apiname); stubws->apiset = afb_apiset_addref(apiset); - stubws->refcount = 1; - afb_proto_ws_on_hangup(stubws->proto, on_hangup); 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->listener = afb_evt_listener_create(&server_evt_itf, stubws); + stubws->cred = afb_cred_create_for_socket(fdev_fd(fdev)); + stubws->listener = afb_evt_listener_create(&server_event_itf, stubws); if (stubws->listener != NULL) return stubws; afb_stub_ws_unref(stubws); @@ -648,11 +711,15 @@ struct afb_stub_ws *afb_stub_ws_create_server(int fd, const char *apiname, struc 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); - afb_proto_ws_unref(stubws->proto); - afb_cred_unref(stubws->cred); + if (stubws && !__atomic_sub_fetch(&stubws->refcount, 1, __ATOMIC_RELAXED)) { + + if (stubws->is_client) { + stubws->robust.reopen = NULL; + if (stubws->robust.release) + stubws->robust.release(stubws->robust.closure); + } + + disconnect(stubws); afb_apiset_unref(stubws->apiset); free(stubws); } @@ -663,7 +730,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,14 +740,14 @@ 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 */ + assert(stubws->is_client); /* check client */ api.closure = stubws; - api.itf = &ws_api_itf; - api.group = NULL; + api.itf = &client_api_itf; + api.group = stubws; /* serialize for reconnections */ return api; } @@ -689,3 +756,14 @@ int afb_stub_ws_client_add(struct afb_stub_ws *stubws, struct afb_apiset *apiset return afb_apiset_add(apiset, stubws->apiname, afb_stub_ws_client_api(stubws)); } +void afb_stub_ws_client_robustify(struct afb_stub_ws *stubws, struct fdev *(*reopen)(void*), void *closure, void (*release)(void*)) +{ + assert(stubws->is_client); /* check client */ + + if (stubws->robust.release) + stubws->robust.release(stubws->robust.closure); + + stubws->robust.reopen = reopen; + stubws->robust.closure = closure; + stubws->robust.release = release; +}