X-Git-Url: https://gerrit.automotivelinux.org/gerrit/gitweb?a=blobdiff_plain;f=src%2Fafb-xreq.c;h=d021c52860399619bb5d9d936c3ffe67bcb13cd3;hb=46ac3bf9554a75e137d5209d27160c1137d42302;hp=ec3a8fe2b62c4dbe1959794714be10b53edc9735;hpb=d0c20134a2c9ecb53045b63cb1e0d6e2aa2b26d2;p=src%2Fapp-framework-binder.git diff --git a/src/afb-xreq.c b/src/afb-xreq.c index ec3a8fe2..d021c528 100644 --- a/src/afb-xreq.c +++ b/src/afb-xreq.c @@ -49,6 +49,8 @@ static inline void xreq_addref(struct afb_xreq *xreq) static inline void xreq_unref(struct afb_xreq *xreq) { if (!__atomic_sub_fetch(&xreq->refcount, 1, __ATOMIC_RELAXED)) { + if (!xreq->replied) + afb_xreq_fail(xreq, "error", "no reply"); if (xreq->hookflags) afb_hook_xreq_end(xreq); xreq->queryitf->unref(xreq); @@ -57,6 +59,16 @@ static inline void xreq_unref(struct afb_xreq *xreq) /******************************************************************************/ +extern const struct afb_req_itf xreq_itf; +extern const struct afb_req_itf xreq_hooked_itf; + +static inline struct afb_req to_req(struct afb_xreq *xreq) +{ + return (struct afb_req){ .itf = xreq->hookflags ? &xreq_hooked_itf : &xreq_itf, .closure = xreq }; +} + +/******************************************************************************/ + struct subcall { struct afb_xreq xreq; @@ -70,8 +82,11 @@ struct subcall int status; }; struct { + union { void (*callback)(void*, int, struct json_object*); - void *closure; + void (*callback2)(void*, int, struct json_object*, struct afb_req); + }; + void *closure; } hooked; }; }; @@ -363,6 +378,32 @@ static void xreq_subcall_cb(void *closure, const char *api, const char *verb, st } } +static void xreq_subcall_req_reply_cb(void *closure, int status, struct json_object *result) +{ + struct subcall *subcall = closure; + subcall->hooked.callback2(subcall->hooked.closure, status, result, to_req(subcall->caller)); +} + +static void xreq_subcall_req_cb(void *closure, const char *api, const char *verb, struct json_object *args, void (*callback)(void*, int, struct json_object*, struct afb_req), void *cb_closure) +{ + struct afb_xreq *xreq = closure; + struct subcall *subcall; + + subcall = subcall_alloc(xreq, api, verb, args); + if (subcall == NULL) { + if (callback) + callback(cb_closure, 1, afb_msg_json_internal_error(), to_req(xreq)); + json_object_put(args); + } else { + subcall->callback = xreq_subcall_req_reply_cb; + subcall->closure = subcall; + subcall->hooked.callback2 = callback; + subcall->hooked.closure = cb_closure; + subcall_process(subcall); + } +} + + static int xreq_subcallsync_cb(void *closure, const char *api, const char *verb, struct json_object *args, struct json_object **result) { int rc; @@ -506,7 +547,7 @@ static int xreq_hooked_unsubscribe_cb(void *closure, struct afb_event event) static void xreq_hooked_subcall_reply_cb(void *closure, int status, struct json_object *result) { struct subcall *subcall = closure; - + afb_hook_xreq_subcall_result(subcall->caller, status, result); subcall->hooked.callback(subcall->hooked.closure, status, result); } @@ -531,6 +572,34 @@ static void xreq_hooked_subcall_cb(void *closure, const char *api, const char *v } } +static void xreq_hooked_subcall_req_reply_cb(void *closure, int status, struct json_object *result) +{ + struct subcall *subcall = closure; + + afb_hook_xreq_subcall_req_result(subcall->caller, status, result); + subcall->hooked.callback2(subcall->hooked.closure, status, result, to_req(subcall->caller)); +} + +static void xreq_hooked_subcall_req_cb(void *closure, const char *api, const char *verb, struct json_object *args, void (*callback)(void*, int, struct json_object*, struct afb_req), void *cb_closure) +{ + struct afb_xreq *xreq = closure; + struct subcall *subcall; + + afb_hook_xreq_subcall_req(xreq, api, verb, args); + subcall = subcall_alloc(xreq, api, verb, args); + if (subcall == NULL) { + if (callback) + callback(cb_closure, 1, afb_msg_json_internal_error(), to_req(xreq)); + json_object_put(args); + } else { + subcall->callback = xreq_hooked_subcall_req_reply_cb; + subcall->closure = subcall; + subcall->hooked.callback2 = callback; + subcall->hooked.closure = cb_closure; + subcall_process(subcall); + } +} + static int xreq_hooked_subcallsync_cb(void *closure, const char *api, const char *verb, struct json_object *args, struct json_object **result) { int r; @@ -578,7 +647,8 @@ const struct afb_req_itf xreq_itf = { .subcall = xreq_subcall_cb, .subcallsync = xreq_subcallsync_cb, .vverbose = xreq_vverbose_cb, - .store = xreq_store_cb + .store = xreq_store_cb, + .subcall_req = xreq_subcall_req_cb }; const struct afb_req_itf xreq_hooked_itf = { @@ -599,14 +669,10 @@ const struct afb_req_itf xreq_hooked_itf = { .subcall = xreq_hooked_subcall_cb, .subcallsync = xreq_hooked_subcallsync_cb, .vverbose = xreq_hooked_vverbose_cb, - .store = xreq_hooked_store_cb + .store = xreq_hooked_store_cb, + .subcall_req = xreq_hooked_subcall_req_cb }; -static inline struct afb_req to_req(struct afb_xreq *xreq) -{ - return (struct afb_req){ .itf = xreq->hookflags ? &xreq_hooked_itf : &xreq_itf, .closure = xreq }; -} - /******************************************************************************/ struct afb_req afb_xreq_unstore(struct afb_stored_req *sreq) @@ -810,7 +876,7 @@ void afb_xreq_fail_unknown_verb(struct afb_xreq *xreq) static void process_sync(struct afb_xreq *xreq) { - struct afb_api api; + const struct afb_api *api; /* init hooking */ afb_hook_init_xreq(xreq); @@ -818,14 +884,15 @@ static void process_sync(struct afb_xreq *xreq) afb_hook_xreq_begin(xreq); /* search the api */ - if (afb_apiset_get_started(xreq->apiset, xreq->api, &api) < 0) { + api = (const struct afb_api*)xreq->context.api_key; + if (api) + api->itf->call(api->closure, xreq); + else { + api = afb_apiset_lookup_started(xreq->apiset, xreq->api, 1); if (errno == ENOENT) afb_xreq_fail_f(xreq, "unknown-api", "api %s not found", xreq->api); else afb_xreq_fail_f(xreq, "bad-api-state", "api %s not started correctly: %m", xreq->api); - } else { - xreq->context.api_key = api.closure; - api.itf->call(api.closure, xreq); } } @@ -843,10 +910,14 @@ static void process_async(int signum, void *arg) void afb_xreq_process(struct afb_xreq *xreq, struct afb_apiset *apiset) { + const struct afb_api *api; + xreq->apiset = apiset; + api = afb_apiset_lookup_started(apiset, xreq->api, 1); + xreq->context.api_key = (void*)api; xreq_addref(xreq); - if (jobs_queue(NULL, afb_apiset_timeout_get(apiset), process_async, xreq) < 0) { + if (jobs_queue(api && api->noconcurrency ? (void*)api : NULL, afb_apiset_timeout_get(apiset), process_async, xreq) < 0) { /* TODO: allows or not to proccess it directly as when no threading? (see above) */ ERROR("can't process job with threads: %m"); afb_xreq_fail_f(xreq, "cancelled", "not able to create a job for the task");