#define _GNU_SOURCE
-#define NO_PLUGIN_VERBOSE_MACRO
-
#include <stdlib.h>
#include <string.h>
#include <assert.h>
#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;
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 */
};
/* 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 **********************************/
{
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;
}
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)
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 */
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 */
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 */
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 */
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 */
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)
/* 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(
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 **********************************/
{
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;
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);
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;
+}
Code Review / src / app-framework-binder.git / blobdiff