X-Git-Url: https://gerrit.automotivelinux.org/gerrit/gitweb?a=blobdiff_plain;f=src%2Fafb-proto-ws.c;h=2781ff092bd6fd93be46807728e8b0593fad2f70;hb=043c27c3a8fd323d59e41288b6fd24f0e9bfa9a3;hp=7644a96aa1a3f447d50cda90e28275ccb555939e;hpb=12ec841c28f8f795b49466cc377e64db3146430d;p=src%2Fapp-framework-binder.git

diff --git a/src/afb-proto-ws.c b/src/afb-proto-ws.c
index 7644a96a..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 <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;
 
@@ -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 */
 };
 
@@ -158,6 +158,9 @@ struct afb_proto_ws
 
 	/* on hangup callback */
 	void (*on_hangup)(void *closure);
+
+	/* queuing facility for processing messages */
+	int (*queuing)(struct afb_proto_ws *proto, void (*process)(int s, void *c), void *closure);
 };
 
 /******************* streaming objects **********************************/
@@ -258,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;
 	}
@@ -325,6 +329,32 @@ static int writebuf_object(struct writebuf *wb, struct json_object *object)
 	return string != NULL && writebuf_string(wb, string);
 }
 
+/******************* queuing of messages *****************/
+
+/* queue the processing of the received message (except if size=0 cause it's not a valid message) */
+static void queue_message_processing(struct afb_proto_ws *protows, char *data, size_t size, void (*processing)(int,void*))
+{
+	struct binary *binary;
+
+	if (size) {
+		binary = malloc(sizeof *binary);
+		if (!binary) {
+			/* TODO process the problem */
+			errno = ENOMEM;
+		} else {
+			binary->protows = protows;
+			binary->rb.base = data;
+			binary->rb.head = data;
+			binary->rb.end = data + size;
+			if (!protows->queuing
+			 || protows->queuing(protows, processing, binary) < 0)
+				processing(0, binary);
+			return;
+		}
+	}
+	free(data);
+}
+
 /******************* ws request part for server *****************/
 
 void afb_proto_ws_call_addref(struct afb_proto_ws_call *call)
@@ -485,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 */
@@ -495,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 */
@@ -506,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 */
@@ -517,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 */
@@ -527,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 */
@@ -538,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)
@@ -624,25 +666,9 @@ static void client_on_binary_job(int sig, void *closure)
 /* callback when receiving binary data */
 static void client_on_binary(void *closure, char *data, size_t size)
 {
-	int rc;
-	struct binary *binary;
+	struct afb_proto_ws *protows = closure;
 
-	if (size) {
-		binary = malloc(sizeof *binary);
-		if (!binary) {
-			errno = ENOMEM;
-		} else {
-			binary->protows = closure;
-			binary->rb.base = data;
-			binary->rb.head = data;
-			binary->rb.end = data + size;
-			rc = jobs_queue(NULL, 0, client_on_binary_job, binary);
-			if (rc >= 0)
-				return;
-			free(binary);
-		}
-	}
-	free(data);
+	queue_message_processing(protows, data, size, client_on_binary_job);
 }
 
 int afb_proto_ws_client_call(
@@ -870,25 +896,9 @@ static void server_on_binary_job(int sig, void *closure)
 
 static void server_on_binary(void *closure, char *data, size_t size)
 {
-	int rc;
-	struct binary *binary;
+	struct afb_proto_ws *protows = closure;
 
-	if (size) {
-		binary = malloc(sizeof *binary);
-		if (!binary) {
-			errno = ENOMEM;
-		} else {
-			binary->protows = closure;
-			binary->rb.base = data;
-			binary->rb.head = data;
-			binary->rb.end = data + size;
-			rc = jobs_queue(NULL, 0, server_on_binary_job, binary);
-			if (rc >= 0)
-				return;
-			free(binary);
-		}
-	}
-	free(data);
+	queue_message_processing(protows, data, size, server_on_binary_job);
 }
 
 /******************* server part: manage events **********************************/
@@ -938,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;
@@ -1014,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);
@@ -1047,3 +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)(struct afb_proto_ws*, void (*)(int,void*), void*))
+{
+	protows->queuing = queuing;
+}