X-Git-Url: https://gerrit.automotivelinux.org/gerrit/gitweb?a=blobdiff_plain;f=src%2Fafb-proto-ws.c;h=3c26205990a261bde50d977b3234095d5d8480ff;hb=6f13ad1989875b5a0ce50b24211fd1fba093735f;hp=86112a323eea2349f0b7740c06af7be116e72920;hpb=7b6940f1524cac6172e71529a989424ff18fb850;p=src%2Fapp-framework-binder.git

diff --git a/src/afb-proto-ws.c b/src/afb-proto-ws.c
index 86112a32..3c262059 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 <jose.bollo@iot.bzh>
  *
  * 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;
 
@@ -132,9 +132,6 @@ struct afb_proto_ws
 	/* count of references */
 	int refcount;
 
-	/* file descriptor */
-	struct fdev *fdev;
-
 	/* resource control */
 	pthread_mutex_t mutex;
 
@@ -160,7 +157,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 +344,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 +512,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 +524,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 +537,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 +550,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 +576,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 +907,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 +934,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;
 }
 
 /*****************************************************/
@@ -955,9 +982,9 @@ static void on_hangup(void *closure)
 		free(cd);
 	}
 
-	if (protows->fdev) {
-		fdev_unref(protows->fdev);
-		protows->fdev = 0;
+	if (protows->ws) {
+		afb_ws_destroy(protows->ws);
+		protows->ws = 0;
 		if (protows->on_hangup)
 			protows->on_hangup(protows->closure);
 	}
@@ -997,7 +1024,6 @@ static struct afb_proto_ws *afb_proto_ws_create(struct fdev *fdev, const struct
 		fcntl(fdev_fd(fdev), F_SETFL, O_NONBLOCK);
 		protows->ws = afb_ws_create(fdev, itf, protows);
 		if (protows->ws != NULL) {
-			protows->fdev = fdev;
 			protows->refcount = 1;
 			protows->closure = closure;
 			protows->server_itf = itfs;
@@ -1024,7 +1050,6 @@ void afb_proto_ws_unref(struct afb_proto_ws *protows)
 {
 	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);
 		free(protows);
 	}
@@ -1047,7 +1072,8 @@ int afb_proto_ws_is_server(struct afb_proto_ws *protows)
 
 void afb_proto_ws_hangup(struct afb_proto_ws *protows)
 {
-	afb_ws_hangup(protows->ws);
+	if (protows->ws)
+		afb_ws_hangup(protows->ws);
 }
 
 void afb_proto_ws_on_hangup(struct afb_proto_ws *protows, void (*on_hangup)(void *closure))
@@ -1055,7 +1081,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;
 }