Tag broadcasted events with UUID and hop

[src/app-framework-binder.git] / src / afb-proto-ws.c

diff --git a/src/afb-proto-ws.c b/src/afb-proto-ws.c
index c079bf6..ea95110 100644 (file)
--- a/src/afb-proto-ws.c
+++ b/src/afb-proto-ws.c
@@ -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 **********************************/
@@ -347,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;
                }
@@ -515,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 */
@@ -525,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 */
@@ -536,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 */
@@ -547,16 +553,21 @@ 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 */
 static void client_on_event_broadcast(struct afb_proto_ws *protows, struct readbuf *rb)
 {
-       const char *event_name;
+       const char *event_name, *uuid;
+       char hop;
        struct json_object *object;
 
-       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);
+       if (protows->client_itf->on_event_broadcast && readbuf_string(rb, &event_name, NULL) && readbuf_object(rb, &object) && (uuid = readbuf_get(rb, 16)) && readbuf_char(rb, &hop))
+               protows->client_itf->on_event_broadcast(protows->closure, event_name, object, (unsigned char*)uuid, (uint8_t)hop);
+       else
+               ERROR("Ignoring broadcast of event");
 }
 
 /* pushs an event */
@@ -568,6 +579,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)
@@ -897,9 +910,9 @@ static int server_event_send(struct afb_proto_ws *protows, char order, const cha
        int rc;
 
        if (writebuf_char(&wb, order)
-        && (order == CHAR_FOR_EVT_BROADCAST || writebuf_uint32(&wb, event_id))
+        && writebuf_uint32(&wb, event_id)
         && writebuf_string(&wb, event_name)
-        && (order == CHAR_FOR_EVT_ADD || order == CHAR_FOR_EVT_DEL || writebuf_object(&wb, data))) {
+        && (order != CHAR_FOR_EVT_PUSH || writebuf_object(&wb, data))) {
                pthread_mutex_lock(&protows->mutex);
                rc = afb_ws_binary_v(protows->ws, wb.iovec, wb.count);
                pthread_mutex_unlock(&protows->mutex);
@@ -924,9 +937,26 @@ int afb_proto_ws_server_event_push(struct afb_proto_ws *protows, const char *eve
        return server_event_send(protows, CHAR_FOR_EVT_PUSH, event_name, event_id, data);
 }
 
-int afb_proto_ws_server_event_broadcast(struct afb_proto_ws *protows, const char *event_name, struct json_object *data)
+int afb_proto_ws_server_event_broadcast(struct afb_proto_ws *protows, const char *event_name, struct json_object *data, const unsigned char uuid[16], uint8_t hop)
 {
-       return server_event_send(protows, CHAR_FOR_EVT_BROADCAST, event_name, 0, data);
+       struct writebuf wb = { .count = 0 };
+       int rc;
+
+       if (!hop--)
+               return 0;
+
+       if (writebuf_char(&wb, CHAR_FOR_EVT_BROADCAST)
+        && writebuf_string(&wb, event_name)
+        && writebuf_object(&wb, data)
+        && writebuf_put(&wb, uuid, 16)
+        && writebuf_char(&wb, (char)hop)) {
+               pthread_mutex_lock(&protows->mutex);
+               rc = afb_ws_binary_v(protows->ws, wb.iovec, wb.count);
+               pthread_mutex_unlock(&protows->mutex);
+               if (rc >= 0)
+                       return 0;
+       }
+       return -1;
 }
 
 /*****************************************************/
@@ -936,8 +966,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;
@@ -1012,7 +1052,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);
@@ -1045,7 +1085,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;
 }