X-Git-Url: https://gerrit.automotivelinux.org/gerrit/gitweb?a=blobdiff_plain;f=src%2Fafb-proto-ws.c;h=2781ff092bd6fd93be46807728e8b0593fad2f70;hb=5715fd569d6f8258830bb1808f9b65bde72cc5cd;hp=142afa98b5ed4a58ffae298ad3171bd82e5e29db;hpb=95ad0012182b5f32cc1dd9843304b58d263a7de0;p=src%2Fapp-framework-binder.git diff --git a/src/afb-proto-ws.c b/src/afb-proto-ws.c index 142afa98..2781ff09 100644 --- a/src/afb-proto-ws.c +++ b/src/afb-proto-ws.c @@ -1,5 +1,5 @@ /* - * Copyright (C) 2015-2018 "IoT.bzh" + * Copyright (C) 2015-2019 "IoT.bzh" * Author José Bollo * * Licensed under the Apache License, Version 2.0 (the "License"); @@ -35,8 +35,8 @@ #include "afb-ws.h" #include "afb-msg-json.h" #include "afb-proto-ws.h" -#include "jobs.h" #include "fdev.h" +#include "verbose.h" struct afb_proto_ws; @@ -89,7 +89,7 @@ For the purpose of handling events the server can: struct client_call { struct client_call *next; /* the next call */ struct afb_proto_ws *protows; /* the proto_ws */ - void *request; + void *request; /* the request closure */ uint32_t callid; /* the message identifier */ }; @@ -160,7 +160,7 @@ struct afb_proto_ws void (*on_hangup)(void *closure); /* queuing facility for processing messages */ - int (*queuing)(void (*process)(int s, void *c), void *closure); + int (*queuing)(struct afb_proto_ws *proto, void (*process)(int s, void *c), void *closure); }; /******************* streaming objects **********************************/ @@ -261,10 +261,11 @@ static int readbuf_object(struct readbuf *rb, struct json_object **object) { const char *string; struct json_object *o; + enum json_tokener_error jerr; int rc = readbuf_string(rb, &string, NULL); if (rc) { - o = json_tokener_parse(string); - if (o == NULL && strcmp(string, "null")) + o = json_tokener_parse_verbose(string, &jerr); + if (jerr != json_tokener_success) o = json_object_new_string(string); *object = o; } @@ -346,7 +347,7 @@ static void queue_message_processing(struct afb_proto_ws *protows, char *data, s binary->rb.head = data; binary->rb.end = data + size; if (!protows->queuing - || protows->queuing(processing, binary) < 0) + || protows->queuing(protows, processing, binary) < 0) processing(0, binary); return; } @@ -514,6 +515,8 @@ static void client_on_event_create(struct afb_proto_ws *protows, struct readbuf if (protows->client_itf->on_event_create && client_msg_event_read(rb, &event_id, &event_name)) protows->client_itf->on_event_create(protows->closure, event_name, (int)event_id); + else + ERROR("Ignoring creation of event"); } /* removes an event */ @@ -524,6 +527,8 @@ static void client_on_event_remove(struct afb_proto_ws *protows, struct readbuf if (protows->client_itf->on_event_remove && client_msg_event_read(rb, &event_id, &event_name)) protows->client_itf->on_event_remove(protows->closure, event_name, (int)event_id); + else + ERROR("Ignoring deletion of event"); } /* subscribes an event */ @@ -535,6 +540,8 @@ static void client_on_event_subscribe(struct afb_proto_ws *protows, struct readb if (protows->client_itf->on_event_subscribe && client_msg_call_get(protows, rb, &call) && client_msg_event_read(rb, &event_id, &event_name)) protows->client_itf->on_event_subscribe(protows->closure, call->request, event_name, (int)event_id); + else + ERROR("Ignoring subscription to event"); } /* unsubscribes an event */ @@ -546,6 +553,8 @@ static void client_on_event_unsubscribe(struct afb_proto_ws *protows, struct rea if (protows->client_itf->on_event_unsubscribe && client_msg_call_get(protows, rb, &call) && client_msg_event_read(rb, &event_id, &event_name)) protows->client_itf->on_event_unsubscribe(protows->closure, call->request, event_name, (int)event_id); + else + ERROR("Ignoring unsubscription to event"); } /* receives broadcasted events */ @@ -556,6 +565,8 @@ static void client_on_event_broadcast(struct afb_proto_ws *protows, struct readb if (protows->client_itf->on_event_broadcast && readbuf_string(rb, &event_name, NULL) && readbuf_object(rb, &object)) protows->client_itf->on_event_broadcast(protows->closure, event_name, object); + else + ERROR("Ignoring broadcast of event"); } /* pushs an event */ @@ -567,6 +578,8 @@ static void client_on_event_push(struct afb_proto_ws *protows, struct readbuf *r if (protows->client_itf->on_event_push && client_msg_event_read(rb, &event_id, &event_name) && readbuf_object(rb, &object)) protows->client_itf->on_event_push(protows->closure, event_name, (int)event_id, object); + else + ERROR("Ignoring push of event"); } static void client_on_reply(struct afb_proto_ws *protows, struct readbuf *rb) @@ -935,8 +948,18 @@ static void on_hangup(void *closure) { struct afb_proto_ws *protows = closure; struct client_describe *cd, *ncd; + struct client_call *call, *ncall; + + ncd = __atomic_exchange_n(&protows->describes, NULL, __ATOMIC_RELAXED); + ncall = __atomic_exchange_n(&protows->calls, NULL, __ATOMIC_RELAXED); + + while (ncall) { + call= ncall; + ncall = call->next; + protows->client_itf->on_reply(protows->closure, call->request, NULL, "disconnected", "server hung up"); + free(call); + } - ncd = protows->describes; while (ncd) { cd= ncd; ncd = cd->next; @@ -1011,7 +1034,7 @@ struct afb_proto_ws *afb_proto_ws_create_server(struct fdev *fdev, const struct void afb_proto_ws_unref(struct afb_proto_ws *protows) { - if (!__atomic_sub_fetch(&protows->refcount, 1, __ATOMIC_RELAXED)) { + if (protows && !__atomic_sub_fetch(&protows->refcount, 1, __ATOMIC_RELAXED)) { afb_proto_ws_hangup(protows); afb_ws_destroy(protows->ws); pthread_mutex_destroy(&protows->mutex); @@ -1044,7 +1067,7 @@ void afb_proto_ws_on_hangup(struct afb_proto_ws *protows, void (*on_hangup)(void protows->on_hangup = on_hangup; } -void afb_proto_ws_set_queuing(struct afb_proto_ws *protows, int (*queuing)(void (*)(int,void*), void*)) +void afb_proto_ws_set_queuing(struct afb_proto_ws *protows, int (*queuing)(struct afb_proto_ws*, void (*)(int,void*), void*)) { protows->queuing = queuing; }