2 files changed, 273 insertions, 59 deletions

diff --git a/src/afb-api-ws.c b/src/afb-api-ws.c
index 9b1d8cc7..ea0781df 100644
--- a/src/afb-api-ws.c
+++ b/src/afb-api-ws.c
@@ -29,6 +29,7 @@
 #include <sys/types.h>
 #include <sys/socket.h>
 #include <sys/un.h>
+#include <pthread.h>
 
 #include <json-c/json.h>
 #include <systemd/sd-event.h>
@@ -62,6 +63,8 @@ struct api_ws_client;
 #define CHAR_FOR_EVT_PUSH         '!'
 #define CHAR_FOR_EVT_SUBSCRIBE    'S'
 #define CHAR_FOR_EVT_UNSUBSCRIBE  'U'
+#define CHAR_FOR_SUBCALL_CALL     'B'
+#define CHAR_FOR_SUBCALL_REPLY    'R'
 
 /*
  */
@@ -70,6 +73,7 @@ struct api_ws
 	char *path;		/* path of the object for the API */
 	char *api;		/* api name of the interface */
 	int fd;			/* file descriptor */
+	pthread_mutex_t mutex;	/**< resource control */
 	union {
 		struct {
 			uint32_t id;
@@ -78,8 +82,7 @@ struct api_ws
 			struct api_ws_memo *memos;
 		} client;
 		struct {
-			sd_event_source *listensrc;
-			struct afb_evt_listener *listener; /* listener for broadcasted events */
+			sd_event_source *listensrc; /**< systemd source for server socket */
 		} server;
 	};
 };
@@ -116,6 +119,29 @@ static const struct afb_evt_itf api_ws_server_evt_itf = {
 	.remove = api_ws_server_event_remove
 };
 
+/******************* handling subcalls *****************************/
+
+/**
+ * Structure on server side for recording pending
+ * subcalls.
+ */
+struct api_ws_subcall
+{
+	struct api_ws_subcall *next;	/**< next subcall for the client */
+	uint32_t subcallid;		/**< the subcallid */
+	void (*callback)(void*, int, struct json_object*); /**< callback on completion */
+	void *closure;			/**< closure of the callback */
+};
+
+/**
+ * Structure for sending back replies on client side
+ */
+struct api_ws_reply
+{
+	struct api_ws *apiws;	/**< api descriptor */
+	uint32_t subcallid;	/**< subcallid for the reply */
+};
+
 /******************* client description part for server *****************************/
 
 struct api_ws_client
@@ -126,17 +152,23 @@ struct api_ws_client
 	/* count of references */
 	int refcount;
 
-	/* listener for events */
-	struct afb_evt_listener *listener;
-
 	/* file descriptor */
 	int fd;
 
+	/* resource control */
+	pthread_mutex_t mutex;
+
+	/* listener for events */
+	struct afb_evt_listener *listener;
+
 	/* websocket */
 	struct afb_ws *ws;
 
 	/* credentials */
 	struct afb_cred *cred;
+
+	/* pending subcalls */
+	struct api_ws_subcall *subcalls;
 };
 
 /******************* websocket interface for client part **********************************/
@@ -161,8 +193,6 @@ static const struct afb_ws_itf api_ws_server_ws_itf =
 struct api_ws_server_req {
 	struct afb_xreq xreq;		/* the xreq */
 	struct api_ws_client *client;	/* the client of the request */
-	const char *request;		/* the readen request as string */
-	size_t lenreq;			/* the length of the request */
 	uint32_t msgid;			/* the incoming request msgid */
 };
 
@@ -171,11 +201,13 @@ static void api_ws_server_req_fail_cb(struct afb_xreq *xreq, const char *status,
 static void api_ws_server_req_destroy_cb(struct afb_xreq *xreq);
 static int api_ws_server_req_subscribe_cb(struct afb_xreq *xreq, struct afb_event event);
 static int api_ws_server_req_unsubscribe_cb(struct afb_xreq *xreq, struct afb_event event);
+static void api_ws_server_req_subcall_cb(struct afb_xreq *xreq, const char *api, const char *verb, struct json_object *args, void (*callback)(void*, int, struct json_object*), void *cb_closure);
 
 const struct afb_xreq_query_itf afb_api_ws_xreq_itf = {
 	.success = api_ws_server_req_success_cb,
 	.fail = api_ws_server_req_fail_cb,
 	.unref = api_ws_server_req_destroy_cb,
+	.subcall = api_ws_server_req_subcall_cb,
 	.subscribe = api_ws_server_req_subscribe_cb,
 	.unsubscribe = api_ws_server_req_unsubscribe_cb
 };
@@ -210,6 +242,7 @@ static struct api_ws *api_ws_make(const char *path)
 		errno = EINVAL;
 		goto error2;
 	}
+	pthread_mutex_init(&api->mutex, NULL);
 
 	api->fd = -1;
 	return api;
@@ -358,6 +391,14 @@ static char *api_ws_read_get(struct readbuf *rb, uint32_t length)
 	return before;
 }
 
+static int api_ws_read_char(struct readbuf *rb, char *value)
+{
+	if (rb->head >= rb->end)
+		return 0;
+	*value = *rb->head++;
+	return 1;
+}
+
 static int api_ws_read_uint32(struct readbuf *rb, uint32_t *value)
 {
 	char *after = rb->head + sizeof *value;
@@ -381,9 +422,16 @@ static int api_ws_read_string(struct readbuf *rb, const char **value, size_t *le
 
 static int api_ws_read_object(struct readbuf *rb, struct json_object **object)
 {
-	size_t length;
 	const char *string;
-	return api_ws_read_string(rb, &string, &length) && ((*object = json_tokener_parse(string)) != NULL) == (strcmp(string, "null") != 0);
+	struct json_object *o;
+	int rc = api_ws_read_string(rb, &string, NULL);
+	if (rc) {
+		o = json_tokener_parse(string);
+		if (o == NULL && strcmp(string, "null"))
+			o = json_object_new_string(string);
+		*object = o;
+	}
+	return rc;
 }
 
 static int api_ws_write_put(struct writebuf *wb, const void *value, size_t length)
@@ -438,22 +486,15 @@ static int api_ws_write_object(struct writebuf *wb, struct json_object *object)
 	return string != NULL && api_ws_write_string(wb, string);
 }
 
-
-
-
 /******************* client part **********************************/
 
 /*
  * structure for recording query data
  */
 struct api_ws_memo {
-	struct api_ws_memo *next;		/* the next memo */
+	struct api_ws_memo *next;	/* the next memo */
 	struct api_ws *api;		/* the ws api */
 	struct afb_xreq *xreq;		/* the request handle */
-#if 0
-	struct afb_req req;		/* the request handle */
-	struct afb_context *context;	/* the context of the query */
-#endif
 	uint32_t msgid;			/* the message identifier */
 };
 
@@ -575,7 +616,7 @@ static int api_ws_client_msg_memo_get(struct api_ws *api, struct readbuf *rb, st
 }
 
 /* read a subscrition message */
-static int api_ws_client_msg_subscription_get(struct api_ws *api, struct readbuf *rb, struct api_ws_event **ev, struct api_ws_memo **memo)
+static int api_ws_client_msg_subscription_get(struct api_ws *api, struct readbuf *rb, struct api_ws_memo **memo, struct api_ws_event **ev)
 {
 	return api_ws_client_msg_memo_get(api, rb, memo) && api_ws_client_msg_event_get(api, rb, ev);
 }
@@ -649,7 +690,7 @@ static void api_ws_client_event_subscribe(struct api_ws *api, struct readbuf *rb
 	struct api_ws_event *ev;
 	struct api_ws_memo *memo;
 
-	if (api_ws_client_msg_subscription_get(api, rb, &ev, &memo)) {
+	if (api_ws_client_msg_subscription_get(api, rb, &memo, &ev)) {
 		/* subscribe the request from the event */
 		if (afb_xreq_subscribe(memo->xreq, ev->event) < 0)
 			ERROR("can't subscribe: %m");
@@ -662,7 +703,7 @@ static void api_ws_client_event_unsubscribe(struct api_ws *api, struct readbuf *
 	struct api_ws_event *ev;
 	struct api_ws_memo *memo;
 
-	if (api_ws_client_msg_subscription_get(api, rb, &ev, &memo)) {
+	if (api_ws_client_msg_subscription_get(api, rb, &memo, &ev)) {
 		/* unsubscribe the request from the event */
 		if (afb_xreq_unsubscribe(memo->xreq, ev->event) < 0)
 			ERROR("can't unsubscribe: %m");
@@ -738,39 +779,102 @@ static void api_ws_client_reply_fail(struct api_ws *api, struct readbuf *rb)
 	api_ws_client_memo_destroy(memo);
 }
 
+/* send a subcall reply */
+static void api_ws_client_send_subcall_reply(struct api_ws_reply *reply, int iserror, json_object *object)
+{
+	int rc;
+	struct writebuf wb = { .count = 0 };
+	char ie = (char)!!iserror;
+
+	if (!api_ws_write_char(&wb, CHAR_FOR_SUBCALL_REPLY)
+	 || !api_ws_write_uint32(&wb, reply->subcallid)
+	 || !api_ws_write_char(&wb, ie)
+	 || !api_ws_write_object(&wb, object)) {
+		/* write error ? */
+		return;
+	}
+
+	rc = afb_ws_binary_v(reply->apiws->client.ws, wb.iovec, wb.count);
+	if (rc >= 0)
+		return;
+	ERROR("error while sending subcall reply");
+}
+
+/* callback for subcall reply */
+static void api_ws_client_subcall_reply_cb(void *closure, int iserror, json_object *object)
+{
+	api_ws_client_send_subcall_reply(closure, iserror, object);
+	free(closure);
+}
+
+/* received a subcall request */
+static void api_ws_client_subcall(struct api_ws *apiws, struct readbuf *rb)
+{
+	struct api_ws_reply *reply;
+	struct api_ws_memo *memo;
+	const char *api, *verb;
+	uint32_t subcallid;
+	struct json_object *object;
+
+	reply = malloc(sizeof *reply);
+	if (!reply)
+		return;
+
+	/* retrieve the message data */
+	if (!api_ws_client_msg_memo_get(apiws, rb, &memo))
+		return;
+
+	if (api_ws_read_uint32(rb, &subcallid)
+	 && api_ws_read_string(rb, &api, NULL)
+	 && api_ws_read_string(rb, &verb, NULL)
+	 && api_ws_read_object(rb, &object)) {
+		reply->apiws = apiws;
+		reply->subcallid = subcallid;
+		afb_xreq_subcall(memo->xreq, api, verb, object, api_ws_client_subcall_reply_cb, reply);
+	}
+}
+
 /* callback when receiving binary data */
 static void api_ws_client_on_binary(void *closure, char *data, size_t size)
 {
 	if (size > 0) {
+		struct api_ws *apiws = closure;
 		struct readbuf rb = { .head = data, .end = data + size };
+
+		pthread_mutex_lock(&apiws->mutex);
 		switch (*rb.head++) {
 		case CHAR_FOR_ANSWER_SUCCESS: /* success */
-			api_ws_client_reply_success(closure, &rb);
+			api_ws_client_reply_success(apiws, &rb);
 			break;
 		case CHAR_FOR_ANSWER_FAIL: /* fail */
-			api_ws_client_reply_fail(closure, &rb);
+			api_ws_client_reply_fail(apiws, &rb);
 			break;
 		case CHAR_FOR_EVT_BROADCAST: /* broadcast */
-			api_ws_client_event_broadcast(closure, &rb);
+			api_ws_client_event_broadcast(apiws, &rb);
 			break;
 		case CHAR_FOR_EVT_ADD: /* creates the event */
-			api_ws_client_event_create(closure, &rb);
+			api_ws_client_event_create(apiws, &rb);
 			break;
 		case CHAR_FOR_EVT_DEL: /* drops the event */
-			api_ws_client_event_drop(closure, &rb);
+			api_ws_client_event_drop(apiws, &rb);
 			break;
 		case CHAR_FOR_EVT_PUSH: /* pushs the event */
-			api_ws_client_event_push(closure, &rb);
+			api_ws_client_event_push(apiws, &rb);
 			break;
 		case CHAR_FOR_EVT_SUBSCRIBE: /* subscribe event for a request */
-			api_ws_client_event_subscribe(closure, &rb);
+			api_ws_client_event_subscribe(apiws, &rb);
 			break;
 		case CHAR_FOR_EVT_UNSUBSCRIBE: /* unsubscribe event for a request */
-			api_ws_client_event_unsubscribe(closure, &rb);
+			api_ws_client_event_unsubscribe(apiws, &rb);
+			break;
+		case CHAR_FOR_SUBCALL_CALL: /* subcall */
+			api_ws_client_subcall(apiws, &rb);
 			break;
 		default: /* unexpected message */
+			/* TODO: close the connection */
 			break;
 		}
+		pthread_mutex_unlock(&apiws->mutex);
 	}
 	free(data);
 }
@@ -783,13 +887,15 @@ static void api_ws_client_call_cb(void * closure, struct afb_xreq *xreq)
 	struct writebuf wb = { .count = 0 };
 	const char *raw;
 	size_t szraw;
-	struct api_ws *api = closure;
+	struct api_ws *apiws = closure;
+
+	pthread_mutex_lock(&apiws->mutex);
 
 	/* create the recording data */
-	memo = api_ws_client_memo_make(api, xreq);
+	memo = api_ws_client_memo_make(apiws, xreq);
 	if (memo == NULL) {
 		afb_xreq_fail_f(xreq, "error", "out of memory");
-		return;
+		goto end;
 	}
 
 	/* creates the call message */
@@ -805,12 +911,10 @@ static void api_ws_client_call_cb(void * closure, struct afb_xreq *xreq)
 		goto overflow;
 
 	/* send */
-	rc = afb_ws_binary_v(api->client.ws, wb.iovec, wb.count);
-	if (rc < 0)
-		goto ws_send_error;
-	return;
+	rc = afb_ws_binary_v(apiws->client.ws, wb.iovec, wb.count);
+	if (rc >= 0)
+		goto end;
 
-ws_send_error:
 	afb_xreq_fail(xreq, "error", "websocket sending error");
 	goto clean_memo;
 
@@ -823,6 +927,8 @@ overflow:
 
 clean_memo:
 	api_ws_client_memo_destroy(memo);
+end:
+	pthread_mutex_unlock(&apiws->mutex);
 }
 
 static int api_ws_service_start_cb(void *closure, int share_session, int onneed)
@@ -912,63 +1018,114 @@ error:
 
 static void api_ws_server_client_unref(struct api_ws_client *client)
 {
-	if (!--client->refcount) {
+	struct api_ws_subcall *sc, *nsc;
+
+	if (!__atomic_sub_fetch(&client->refcount, 1, __ATOMIC_RELAXED)) {
 		afb_evt_listener_unref(client->listener);
 		afb_ws_destroy(client->ws);
+		nsc = client->subcalls;
+		while (nsc) {
+			sc= nsc;
+			nsc = sc->next;
+			sc->callback(sc->closure, 1, NULL);
+			free(sc);
+		}
 		afb_cred_unref(client->cred);
 		free(client);
 	}
 }
 
+static void api_ws_server_client_addref(struct api_ws_client *client)
+{
+	__atomic_add_fetch(&client->refcount, 1, __ATOMIC_RELAXED);
+}
+
 /* on call, propagate it to the ws service */
 static void api_ws_server_on_call(struct api_ws_client *client, struct readbuf *rb)
 {
 	struct api_ws_server_req *wreq;
+	char *cverb;
 	const char *uuid, *verb;
-	uint32_t flags;
-
-	client->refcount++;
-
-	/* create the request */
-	wreq = calloc(1 , sizeof *wreq);
-	if (wreq == NULL)
-		goto out_of_memory;
+	uint32_t flags, msgid;
+	size_t lenverb;
+	struct json_object *object;
 
-	wreq->client = client;
+	api_ws_server_client_addref(client);
 
 	/* reads the call message data */
-	if (!api_ws_read_uint32(rb, &wreq->msgid)
+	if (!api_ws_read_uint32(rb, &msgid)
 	 || !api_ws_read_uint32(rb, &flags)
-	 || !api_ws_read_string(rb, &verb, NULL)
+	 || !api_ws_read_string(rb, &verb, &lenverb)
 	 || !api_ws_read_string(rb, &uuid, NULL)
-	 || !api_ws_read_string(rb, &wreq->request, &wreq->lenreq))
+	 || !api_ws_read_object(rb, &object))
 		goto overflow;
 
+	/* create the request */
+	wreq = malloc(++lenverb + sizeof *wreq);
+	if (wreq == NULL)
+		goto out_of_memory;
+
 	afb_xreq_init(&wreq->xreq, &afb_api_ws_xreq_itf);
-	wreq->xreq.json = json_tokener_parse(wreq->request);
-	if (wreq->xreq.json == NULL && strcmp(wreq->request, "null")) {
-		wreq->xreq.json = json_object_new_string(wreq->request);
-	}
+	wreq->client = client;
+	wreq->msgid = msgid;
+	cverb = (char*)&wreq[1];
+	memcpy(cverb, verb, lenverb);
 
 	/* init the context */
 	if (afb_context_connect(&wreq->xreq.context, uuid, NULL) < 0)
-		goto out_of_memory;
+		goto unconnected;
 	wreq->xreq.context.flags = flags;
 
 	/* makes the call */
 	wreq->xreq.cred = afb_cred_addref(client->cred);
 	wreq->xreq.api = client->api;
-	wreq->xreq.verb = verb;
+	wreq->xreq.verb = cverb;
+	wreq->xreq.json = object;
 	afb_apis_call(&wreq->xreq);
 	afb_xreq_unref(&wreq->xreq);
 	return;
 
+unconnected:
+	free(wreq);
 out_of_memory:
+	json_object_put(object);
 overflow:
-	free(wreq);
 	api_ws_server_client_unref(client);
 }
 
+/* on subcall reply */
+static void api_ws_server_on_subcall_reply(struct api_ws_client *client, struct readbuf *rb)
+{
+	char iserror;
+	uint32_t subcallid;
+	struct json_object *object;
+	struct api_ws_subcall *sc, **psc;
+
+	/* reads the call message data */
+	if (!api_ws_read_uint32(rb, &subcallid)
+	 || !api_ws_read_char(rb, &iserror)
+	 || !api_ws_read_object(rb, &object)) {
+		/* TODO bad protocol */
+		return;
+	}
+
+	/* search the subcall and unlink it */
+	pthread_mutex_lock(&client->mutex);
+	psc = &client->subcalls;
+	while ((sc = *psc) && sc->subcallid != subcallid)
+		psc = &sc->next;
+	if (!sc) {
+		pthread_mutex_unlock(&client->mutex);
+		/* TODO subcall not found */
+	} else {
+		*psc = sc->next;
+		pthread_mutex_unlock(&client->mutex);
+		sc->callback(sc->closure, (int)iserror, object);
+		free(sc);
+	}
+	json_object_put(object);
+}
+
 /* callback when receiving binary data */
 static void api_ws_server_on_binary(void *closure, char *data, size_t size)
 {
@@ -978,6 +1135,9 @@ static void api_ws_server_on_binary(void *closure, char *data, size_t size)
 		case CHAR_FOR_CALL:
 			api_ws_server_on_call(closure, &rb);
 			break;
+		case CHAR_FOR_SUBCALL_REPLY:
+			api_ws_server_on_subcall_reply(closure, &rb);
+			break;
 		default: /* unexpected message */
 			/* TODO: close the connection */
 			break;
@@ -1021,6 +1181,7 @@ static void api_ws_server_accept(struct api_ws *api)
 				if (client->ws != NULL) {
 					client->api = api->api;
 					client->refcount = 1;
+					client->subcalls = NULL;
 					return;
 				}
 				afb_cred_unref(client->cred);
@@ -1135,6 +1296,47 @@ static void api_ws_server_req_fail_cb(struct afb_xreq *xreq, const char *status,
 	ERROR("error while sending fail");
 }
 
+static void api_ws_server_req_subcall_cb(struct afb_xreq *xreq, const char *api, const char *verb, struct json_object *args, void (*callback)(void*, int, struct json_object*), void *cb_closure)
+{
+	int rc;
+	struct writebuf wb = { .count = 0 };
+	struct api_ws_subcall *sc, *osc;
+	struct api_ws_server_req *wreq = CONTAINER_OF_XREQ(struct api_ws_server_req, xreq);
+	struct api_ws_client *client = wreq->client;
+
+	sc = malloc(sizeof *sc);
+	if (!sc) {
+
+	} else {
+		sc->callback = callback;
+		sc->closure = cb_closure;
+
+		pthread_mutex_unlock(&client->mutex);
+		sc->subcallid = (uint32_t)(((intptr_t)sc) >> 6);
+		do {
+			sc->subcallid++;
+			osc = client->subcalls;
+			while(osc && osc->subcallid != sc->subcallid)
+				osc = osc->next;
+		} while (osc);
+		sc->next = client->subcalls;
+		client->subcalls = sc;
+		pthread_mutex_unlock(&client->mutex);
+
+		if (api_ws_write_char(&wb, CHAR_FOR_SUBCALL_CALL)
+		 && api_ws_write_uint32(&wb, wreq->msgid)
+		 && api_ws_write_uint32(&wb, sc->subcallid)
+		 && api_ws_write_string(&wb, api)
+		 && api_ws_write_string(&wb, verb)
+		 && api_ws_write_object(&wb, args)) {
+			rc = afb_ws_binary_v(wreq->client->ws, wb.iovec, wb.count);
+			if (rc >= 0)
+				return;
+		}
+		ERROR("error while sending fail");
+	}
+}
+
 static int api_ws_server_req_subscribe_cb(struct afb_xreq *xreq, struct afb_event event)
 {
 	int rc, rc2;
diff --git a/stress-server.sh b/stress-server.sh
index a0d83fd9..31fce9f6 100755
--- a/stress-server.sh
+++ b/stress-server.sh
@@ -3,16 +3,28 @@
 ROOT=$(dirname $0)
 echo ROOT=$ROOT
 
-AFB=$ROOT/build/src/afb-daemon
+cd $ROOT
+
+AFB=build/src/afb-daemon
 HELLO=build/bindings/samples/helloWorld.so
 PORT=12345
 TEST=test
 TOKEN=knock-knock-knoc
-OUT=$ROOT/stress-out-server
+OUT=stress-out-server
 
 rm $OUT*
 
-ARGS="-q --session-max=100 --port=$PORT --workdir=$ROOT --roothttp=$TEST --token=$TOKEN --ldpaths=/tmp --binding=$HELLO"
+case "$1" in
+ --ws)
+  shift
+  ARGS="-q --ldpaths=/tmp --binding=$HELLO --session-max=100 --ws-server=unix:hello --no-httpd --exec $AFB --session-max=100 --port=$PORT --ldpaths=/tmp --roothttp=$TEST --token=$TOKEN --ws-client=unix:hello "
+#  ARGS="-vv --tracereq=all --ldpaths=/tmp --binding=$HELLO --session-max=100 --ws-server=unix:hello --no-httpd --exec $AFB --session-max=100 --port=$PORT --ldpaths=/tmp --roothttp=$TEST --token=$TOKEN --ws-client=unix:hello "
+  ;;
+ *)
+  ARGS="-q --session-max=100 --port=$PORT --workdir=$ROOT --roothttp=$TEST --token=$TOKEN --ldpaths=/tmp --binding=$HELLO"
+  ;;
+esac
+
 
 echo -n launch afb...
 case "$1" in