/*
- * Copyright (C) 2016, 2017, 2018 "IoT.bzh"
+ * Copyright (C) 2015-2020 "IoT.bzh"
* Author: José Bollo <jose.bollo@iot.bzh>
*
* Licensed under the Apache License, Version 2.0 (the "License");
#define RETERR 4
#define EVENT 5
+#define WEBSOCKET_CODE_POLICY_VIOLATION 1008
+#define WEBSOCKET_CODE_INTERNAL_ERROR 1011
+
static void wsj1_on_hangup(struct afb_wsj1 *wsj1);
static void wsj1_on_text(struct afb_wsj1 *wsj1, char *text, size_t size);
+static struct afb_wsj1_msg *wsj1_msg_make(struct afb_wsj1 *wsj1, char *text, size_t size);
static struct afb_ws_itf wsj1_itf = {
.on_hangup = (void*)wsj1_on_hangup,
struct afb_wsj1_msg *next, *previous;
char *text;
int code;
- char *id;
- char *api;
- char *verb;
- char *event;
- char *object_s;
+ const char *id;
+ const char *api;
+ const char *verb;
+ const char *event;
+ const char *object_s;
size_t object_s_length;
- char *token;
+ const char *token;
struct json_object *object_j;
};
static void wsj1_on_hangup(struct afb_wsj1 *wsj1)
{
+ struct wsj1_call *call, *ncall;
+ struct afb_wsj1_msg *msg;
+ char *text;
+ int len;
+
+ static const char error_object_str[] = "{"
+ "\"jtype\":\"afb-reply\","
+ "\"request\":{"
+ "\"status\":\"disconnected\","
+ "\"info\":\"server hung up\"}}";
+
+ ncall = __atomic_exchange_n(&wsj1->calls, NULL, __ATOMIC_RELAXED);
+ while (ncall) {
+ call = ncall;
+ ncall = call->next;
+ len = asprintf(&text, "[%d,\"%s\",%s]", RETERR, call->id, error_object_str);
+ if (len > 0) {
+ msg = wsj1_msg_make(wsj1, text, (size_t)len);
+ if (msg != NULL) {
+ call->callback(call->closure, msg);
+ afb_wsj1_msg_unref(msg);
+ }
+ }
+ free(call);
+ }
+
if (wsj1->itf->on_hangup != NULL)
wsj1->itf->on_hangup(wsj1->closure, wsj1);
}
return wsj1_msg_parse_extract(text, offset, size);
}
-static void wsj1_on_text(struct afb_wsj1 *wsj1, char *text, size_t size)
+static struct afb_wsj1_msg *wsj1_msg_make(struct afb_wsj1 *wsj1, char *text, size_t size)
{
size_t items[10][2];
int n;
struct afb_wsj1_msg *msg;
- struct wsj1_call *call = NULL;
+ char *verb;
/* allocate */
msg = calloc(1, sizeof *msg);
- if (msg == NULL)
+ if (msg == NULL) {
+ errno = ENOMEM;
goto alloc_error;
+ }
/* scan */
n = wsj1_msg_scan(text, items);
- if (n < 0)
+ if (n <= 0)
goto bad_header;
/* scans code: 2|3|4|5 */
- if (items[0][1] != 1) goto bad_header;
+ if (items[0][1] != 1)
+ goto bad_header;
+
switch (text[items[0][0]]) {
case '2': msg->code = CALL; break;
case '3': msg->code = RETOK; break;
if (n != 4 && n != 5) goto bad_header;
msg->id = wsj1_msg_parse_string(text, items[1][0], items[1][1]);
msg->api = wsj1_msg_parse_string(text, items[2][0], items[2][1]);
- msg->verb = strchr(msg->api, '/');
- if (msg->verb == NULL) goto bad_header;
- *msg->verb++ = 0;
+ verb = strchr(msg->api, '/');
+ if (verb == NULL) goto bad_header;
+ *verb++ = 0;
+ if (!*verb || *verb == '/') goto bad_header;
+ msg->verb = verb;
msg->object_s = wsj1_msg_parse_extract(text, items[3][0], items[3][1]);
msg->object_s_length = items[3][1];
msg->token = n == 5 ? wsj1_msg_parse_string(text, items[4][0], items[4][1]) : NULL;
case RETERR:
if (n != 3 && n != 4) goto bad_header;
msg->id = wsj1_msg_parse_string(text, items[1][0], items[1][1]);
- call = wsj1_call_search(wsj1, msg->id, 1);
- if (call == NULL) goto bad_header;
msg->object_s = wsj1_msg_parse_extract(text, items[2][0], items[2][1]);
msg->object_s_length = items[2][1];
msg->token = n == 5 ? wsj1_msg_parse_string(text, items[3][0], items[3][1]) : NULL;
wsj1->messages = msg;
pthread_mutex_unlock(&wsj1->mutex);
- /* incoke the handler */
+ return msg;
+
+bad_header:
+ errno = EBADMSG;
+ free(msg);
+
+alloc_error:
+ free(text);
+ return NULL;
+}
+
+static void wsj1_on_text(struct afb_wsj1 *wsj1, char *text, size_t size)
+{
+ struct wsj1_call *call;
+ struct afb_wsj1_msg *msg;
+
+ /* allocate */
+ msg = wsj1_msg_make(wsj1, text, size);
+ if (msg == NULL) {
+ afb_ws_close(wsj1->ws, errno == EBADMSG
+ ? WEBSOCKET_CODE_POLICY_VIOLATION
+ : WEBSOCKET_CODE_INTERNAL_ERROR, NULL);
+ return;
+ }
+
+ /* handle the message */
switch (msg->code) {
case CALL:
wsj1->itf->on_call(wsj1->closure, msg->api, msg->verb, msg);
break;
case RETOK:
case RETERR:
- call->callback(call->closure, msg);
+ call = wsj1_call_search(wsj1, msg->id, 1);
+ if (call == NULL)
+ afb_ws_close(wsj1->ws, WEBSOCKET_CODE_POLICY_VIOLATION, NULL);
+ else
+ call->callback(call->closure, msg);
free(call);
break;
case EVENT:
break;
}
afb_wsj1_msg_unref(msg);
- return;
-
-bad_header:
- free(msg);
-alloc_error:
- free(text);
- afb_ws_close(wsj1->ws, 1008, NULL);
}
void afb_wsj1_msg_addref(struct afb_wsj1_msg *msg)
struct json_object *afb_wsj1_msg_object_j(struct afb_wsj1_msg *msg)
{
+ enum json_tokener_error jerr;
struct json_object *object = msg->object_j;
if (object == NULL) {
pthread_mutex_lock(&msg->wsj1->mutex);
json_tokener_reset(msg->wsj1->tokener);
- object = json_tokener_parse_ex(msg->wsj1->tokener, msg->object_s, (int)msg->object_s_length);
- if (object == NULL && json_tokener_get_error(msg->wsj1->tokener) == json_tokener_continue)
- object = json_tokener_parse_ex(msg->wsj1->tokener, "", 1);
+ object = json_tokener_parse_ex(msg->wsj1->tokener, msg->object_s, 1 + (int)msg->object_s_length);
+ jerr = json_tokener_get_error(msg->wsj1->tokener);
pthread_mutex_unlock(&msg->wsj1->mutex);
- if (object == NULL) {
+ if (jerr != json_tokener_success) {
/* lazy error detection of json request. Is it to improve? */
object = json_object_new_string_len(msg->object_s, (int)msg->object_s_length);
}
Code Review / src / app-framework-binder.git / blobdiff