#include <json-c/json.h>
-#include "afb-ws.h"
+#include <afb/afb-req-itf.h>
+
+#include "afb-wsj1.h"
#include "afb-ws-json1.h"
+#include "afb-common.h"
#include "afb-msg-json.h"
#include "session.h"
-#include <afb/afb-req-itf.h>
#include "afb-apis.h"
#include "afb-context.h"
+#include "afb-evt.h"
+#include "afb-subcall.h"
+#include "verbose.h"
-static void aws_on_hangup(struct afb_ws_json1 *ws);
-static void aws_on_text(struct afb_ws_json1 *ws, char *text, size_t size);
+/* predeclaration of structures */
+struct afb_ws_json1;
+struct afb_wsreq;
-static struct afb_ws_itf aws_itf = {
- .on_hangup = (void*)aws_on_hangup,
- .on_text = (void*)aws_on_text
-};
+/* predeclaration of websocket callbacks */
+static void aws_on_hangup(struct afb_ws_json1 *ws, struct afb_wsj1 *wsj1);
+static void aws_on_call(struct afb_ws_json1 *ws, const char *api, const char *verb, struct afb_wsj1_msg *msg);
+static void aws_on_event(struct afb_ws_json1 *ws, const char *event, int eventid, struct json_object *object);
-struct afb_wsreq;
+/* predeclaration of wsreq callbacks */
+static void wsreq_addref(struct afb_wsreq *wsreq);
+static void wsreq_unref(struct afb_wsreq *wsreq);
+static struct json_object *wsreq_json(struct afb_wsreq *wsreq);
+static struct afb_arg wsreq_get(struct afb_wsreq *wsreq, const char *name);
+static void wsreq_fail(struct afb_wsreq *wsreq, const char *status, const char *info);
+static void wsreq_success(struct afb_wsreq *wsreq, struct json_object *obj, const char *info);
+static const char *wsreq_raw(struct afb_wsreq *wsreq, size_t *size);
+static void wsreq_send(struct afb_wsreq *wsreq, const char *buffer, size_t size);
+static int wsreq_subscribe(struct afb_wsreq *wsreq, struct afb_event event);
+static int wsreq_unsubscribe(struct afb_wsreq *wsreq, struct afb_event event);
+static void wsreq_subcall(struct afb_wsreq *wsreq, const char *api, const char *verb, struct json_object *args, void (*callback)(void*, int, struct json_object*), void *closure);
+/* declaration of websocket structure */
struct afb_ws_json1
{
+ int refcount;
void (*cleanup)(void*);
void *cleanup_closure;
- struct afb_wsreq *requests;
struct AFB_clientCtx *session;
- struct json_tokener *tokener;
- struct afb_ws *ws;
+ struct afb_evt_listener *listener;
+ struct afb_wsj1 *wsj1;
int new_session;
};
-static void aws_send_event(struct afb_ws_json1 *ws, const char *event, struct json_object *object);
+/* declaration of wsreq structure */
+struct afb_wsreq
+{
+ /*
+ * CAUTION: 'context' field should be the first because there
+ * is an implicit convertion to struct afb_context
+ */
+ struct afb_context context;
+ int refcount;
+ struct afb_ws_json1 *aws;
+ struct afb_wsreq *next;
+ struct afb_wsj1_msg *msgj1;
+};
-static const struct afb_event_listener_itf event_listener_itf = {
- .send = (void*)aws_send_event,
- .expects = NULL
+/* interface for afb_ws_json1 / afb_wsj1 */
+static struct afb_wsj1_itf wsj1_itf = {
+ .on_hangup = (void*)aws_on_hangup,
+ .on_call = (void*)aws_on_call
};
-static inline struct afb_event_listener listener_for(struct afb_ws_json1 *aws)
-{
- return (struct afb_event_listener){ .itf = &event_listener_itf, .closure = aws };
-}
+/* interface for wsreq / afb_req */
+const struct afb_req_itf afb_ws_json1_req_itf = {
+ .json = (void*)wsreq_json,
+ .get = (void*)wsreq_get,
+ .success = (void*)wsreq_success,
+ .fail = (void*)wsreq_fail,
+ .raw = (void*)wsreq_raw,
+ .send = (void*)wsreq_send,
+ .context_get = (void*)afb_context_get,
+ .context_set = (void*)afb_context_set,
+ .addref = (void*)wsreq_addref,
+ .unref = (void*)wsreq_unref,
+ .session_close = (void*)afb_context_close,
+ .session_set_LOA = (void*)afb_context_change_loa,
+ .subscribe = (void*)wsreq_subscribe,
+ .unsubscribe = (void*)wsreq_unsubscribe,
+ .subcall = (void*)wsreq_subcall
+};
+
+/* the interface for events */
+static const struct afb_evt_itf evt_itf = {
+ .broadcast = (void*)aws_on_event,
+ .push = (void*)aws_on_event
+};
+
+/***************************************************************
+****************************************************************
+**
+** functions of afb_ws_json1 / afb_wsj1
+**
+****************************************************************
+***************************************************************/
struct afb_ws_json1 *afb_ws_json1_create(int fd, struct afb_context *context, void (*cleanup)(void*), void *cleanup_closure)
{
if (result == NULL)
goto error;
+ result->refcount = 1;
result->cleanup = cleanup;
result->cleanup_closure = cleanup_closure;
- result->requests = NULL;
result->session = ctxClientAddRef(context->session);
result->new_session = context->created != 0;
if (result->session == NULL)
goto error2;
- result->tokener = json_tokener_new();
- if (result->tokener == NULL)
+ result->wsj1 = afb_wsj1_create(afb_common_get_event_loop(), fd, &wsj1_itf, result);
+ if (result->wsj1 == NULL)
goto error3;
- result->ws = afb_ws_create(fd, &aws_itf, result);
- if (result->ws == NULL)
+ result->listener = afb_evt_listener_create(&evt_itf, result);
+ if (result->listener == NULL)
goto error4;
- if (0 > ctxClientEventListenerAdd(result->session, listener_for(result)))
- goto error5;
-
return result;
-error5:
- afb_ws_destroy(result->ws);
error4:
- json_tokener_free(result->tokener);
+ afb_wsj1_unref(result->wsj1);
error3:
ctxClientUnref(result->session);
error2:
return NULL;
}
-static void aws_on_hangup(struct afb_ws_json1 *ws)
+static struct afb_ws_json1 *aws_addref(struct afb_ws_json1 *ws)
{
- ctxClientEventListenerRemove(ws->session, listener_for(ws));
- afb_ws_destroy(ws->ws);
- json_tokener_free(ws->tokener);
- if (ws->cleanup != NULL)
- ws->cleanup(ws->cleanup_closure);
- ctxClientUnref(ws->session);
- free(ws);
+ ws->refcount++;
+ return ws;
}
-#define CALL 2
-#define RETOK 3
-#define RETERR 4
-#define EVENT 5
-
-struct afb_wsreq
-{
- struct afb_context context;
- int refcount;
- struct afb_ws_json1 *aws;
- struct afb_wsreq *next;
- char *text;
- size_t size;
- int code;
- char *id;
- size_t idlen;
- char *api;
- size_t apilen;
- char *verb;
- size_t verblen;
- char *obj;
- size_t objlen;
- char *tok;
- size_t toklen;
- struct json_object *root;
-};
-
-static void wsreq_addref(struct afb_wsreq *wsreq);
-static void wsreq_unref(struct afb_wsreq *wsreq);
-static struct json_object *wsreq_json(struct afb_wsreq *wsreq);
-static struct afb_arg wsreq_get(struct afb_wsreq *wsreq, const char *name);
-static void wsreq_fail(struct afb_wsreq *wsreq, const char *status, const char *info);
-static void wsreq_success(struct afb_wsreq *wsreq, struct json_object *obj, const char *info);
-static const char *wsreq_raw(struct afb_wsreq *wsreq, size_t *size);
-static void wsreq_send(struct afb_wsreq *wsreq, const char *buffer, size_t size);
-
-
-static const struct afb_req_itf wsreq_itf = {
- .json = (void*)wsreq_json,
- .get = (void*)wsreq_get,
- .success = (void*)wsreq_success,
- .fail = (void*)wsreq_fail,
- .raw = (void*)wsreq_raw,
- .send = (void*)wsreq_send,
- .context_get = (void*)afb_context_get,
- .context_set = (void*)afb_context_set,
- .addref = (void*)wsreq_addref,
- .unref = (void*)wsreq_unref
-};
-
-static int aws_wsreq_parse(struct afb_wsreq *r, char *text, size_t size)
+static void aws_unref(struct afb_ws_json1 *ws)
{
- char *pos, *end, c;
- int aux;
-
- /* scan */
- pos = text;
- end = text + size;
-
- /* scans: [ */
- while(pos < end && *pos == ' ') pos++;
- if (pos == end) goto bad_header;
- if (*pos++ != '[') goto bad_header;
-
- /* scans code: 2|3|4 */
- while(pos < end && *pos == ' ') pos++;
- if (pos == end) goto bad_header;
- switch (*pos++) {
- case '2': r->code = CALL; break;
- case '3': r->code = RETOK; break;
- case '4': r->code = RETERR; break;
- default: goto bad_header;
- }
-
- /* scans: , */
- while(pos < end && *pos == ' ') pos++;
- if (pos == end) goto bad_header;
- if (*pos++ != ',') goto bad_header;
-
- /* scans id: "id" */
- while(pos < end && *pos == ' ') pos++;
- if (pos == end) goto bad_header;
- if (*pos++ != '"') goto bad_header;
- r->id = pos;
- while(pos < end && *pos != '"') pos++;
- if (pos == end) goto bad_header;
- r->idlen = (size_t)(pos++ - r->id);
-
- /* scans: , */
- while(pos < end && *pos == ' ') pos++;
- if (pos == end) goto bad_header;
- if (*pos++ != ',') goto bad_header;
-
- /* scans the method if needed */
- if (r->code == CALL) {
- /* scans: " */
- while(pos < end && *pos == ' ') pos++;
- if (pos == end) goto bad_header;
- if (*pos++ != '"') goto bad_header;
-
- /* scans: api/ */
- r->api = pos;
- while(pos < end && *pos != '"' && *pos != '/') pos++;
- if (pos == end) goto bad_header;
- if (*pos != '/') goto bad_header;
- r->apilen = (size_t)(pos++ - r->api);
- if (r->apilen && r->api[r->apilen - 1] == '\\')
- r->apilen--;
-
- /* scans: verb" */
- r->verb = pos;
- while(pos < end && *pos != '"') pos++;
- if (pos == end) goto bad_header;
- r->verblen = (size_t)(pos++ - r->verb);
-
- /* scans: , */
- while(pos < end && *pos == ' ') pos++;
- if (pos == end) goto bad_header;
- if (*pos++ != ',') goto bad_header;
- }
-
- /* scan obj */
- while(pos < end && *pos == ' ') pos++;
- if (pos == end) goto bad_header;
- aux = 0;
- r->obj = pos;
- while (pos < end && (aux != 0 || (*pos != ',' && *pos != ']'))) {
- if (pos == end) goto bad_header;
- switch(*pos) {
- case '{': case '[': aux++; break;
- case '}': case ']': if (!aux--) goto bad_header; break;
- case '"':
- do {
- pos += 1 + (*pos == '\\');
- } while(pos < end && *pos != '"');
- default:
- break;
- }
- pos++;
- }
- if (pos > end) goto bad_header;
- if (pos == end && aux != 0) goto bad_header;
- c = *pos;
- r->objlen = (size_t)(pos++ - r->obj);
- while (r->objlen && r->obj[r->objlen - 1] == ' ')
- r->objlen--;
-
- /* scan the token (if any) */
- if (c == ',') {
- /* scans token: "token" */
- while(pos < end && *pos == ' ') pos++;
- if (pos == end) goto bad_header;
- if (*pos++ != '"') goto bad_header;
- r->tok = pos;
- while(pos < end && *pos != '"') pos++;
- if (pos == end) goto bad_header;
- r->toklen = (size_t)(pos++ - r->tok);
- while(pos < end && *pos == ' ') pos++;
- if (pos == end) goto bad_header;
- c = *pos++;
+ if (--ws->refcount == 0) {
+ afb_evt_listener_unref(ws->listener);
+ afb_wsj1_unref(ws->wsj1);
+ if (ws->cleanup != NULL)
+ ws->cleanup(ws->cleanup_closure);
+ ctxClientUnref(ws->session);
+ free(ws);
}
+}
- /* scan: ] */
- if (c != ']') goto bad_header;
- while(pos < end && *pos == ' ') pos++;
- if (pos != end) goto bad_header;
-
- /* done */
- r->text = text;
- r->size = size;
- return 1;
-
-bad_header:
- return 0;
+static void aws_on_hangup(struct afb_ws_json1 *ws, struct afb_wsj1 *wsj1)
+{
+ aws_unref(ws);
}
-static void aws_on_text(struct afb_ws_json1 *ws, char *text, size_t size)
+static void aws_on_call(struct afb_ws_json1 *ws, const char *api, const char *verb, struct afb_wsj1_msg *msg)
{
struct afb_req r;
struct afb_wsreq *wsreq;
+ DEBUG("received websocket request for %s/%s: %s", api, verb, afb_wsj1_msg_object_s(msg));
+
/* allocate */
wsreq = calloc(1, sizeof *wsreq);
- if (wsreq == NULL)
- goto alloc_error;
-
- /* init */
- if (!aws_wsreq_parse(wsreq, text, size))
- goto bad_header;
+ if (wsreq == NULL) {
+ afb_wsj1_close(ws->wsj1, 1008, NULL);
+ return;
+ }
- /* fill and record the request */
- if (wsreq->tok != NULL)
- wsreq->tok[wsreq->toklen] = 0;
- afb_context_init(&wsreq->context, ws->session, wsreq->tok);
+ /* init the context */
+ afb_context_init(&wsreq->context, ws->session, afb_wsj1_msg_token(msg));
if (!wsreq->context.invalidated)
wsreq->context.validated = 1;
if (ws->new_session != 0) {
wsreq->context.created = 1;
ws->new_session = 0;
}
+
+ /* fill and record the request */
+ afb_wsj1_msg_addref(msg);
+ wsreq->msgj1 = msg;
wsreq->refcount = 1;
- wsreq->aws = ws;
- wsreq->next = ws->requests;
- ws->requests = wsreq;
+ wsreq->aws = aws_addref(ws);
+ /* emits the call */
r.closure = wsreq;
- r.itf = &wsreq_itf;
- afb_apis_call(r, &wsreq->context, wsreq->api, wsreq->apilen, wsreq->verb, wsreq->verblen);
+ r.itf = &afb_ws_json1_req_itf;
+ afb_apis_call_(r, &wsreq->context, api, verb);
wsreq_unref(wsreq);
- return;
-
-bad_header:
- free(wsreq);
-alloc_error:
- free(text);
- afb_ws_close(ws->ws, 1008, NULL);
- return;
}
+static void aws_on_event(struct afb_ws_json1 *aws, const char *event, int eventid, struct json_object *object)
+{
+ afb_wsj1_send_event_j(aws->wsj1, event, afb_msg_json_event(event, object));
+}
+
+/***************************************************************
+****************************************************************
+**
+** functions of wsreq / afb_req
+**
+****************************************************************
+***************************************************************/
+
static void wsreq_addref(struct afb_wsreq *wsreq)
{
wsreq->refcount++;
static void wsreq_unref(struct afb_wsreq *wsreq)
{
if (--wsreq->refcount == 0) {
- struct afb_wsreq **prv = &wsreq->aws->requests;
- while(*prv != NULL) {
- if (*prv == wsreq) {
- *prv = wsreq->next;
- break;
- }
- prv = &(*prv)->next;
- }
afb_context_disconnect(&wsreq->context);
- json_object_put(wsreq->root);
- free(wsreq->text);
+ afb_wsj1_msg_unref(wsreq->msgj1);
+ aws_unref(wsreq->aws);
free(wsreq);
}
}
static struct json_object *wsreq_json(struct afb_wsreq *wsreq)
{
- struct json_object *root = wsreq->root;
- if (root == NULL) {
- json_tokener_reset(wsreq->aws->tokener);
- root = json_tokener_parse_ex(wsreq->aws->tokener, wsreq->obj, (int)wsreq->objlen);
- if (root == NULL) {
- /* lazy error detection of json request. Is it to improve? */
- root = json_object_new_string_len(wsreq->obj, (int)wsreq->objlen);
- }
- wsreq->root = root;
- }
- return root;
+ return afb_wsj1_msg_object_j(wsreq->msgj1);
}
static struct afb_arg wsreq_get(struct afb_wsreq *wsreq, const char *name)
{
- struct afb_arg arg;
- struct json_object *value, *root;
-
- root = wsreq_json(wsreq);
- if (json_object_object_get_ex(root, name, &value)) {
- arg.name = name;
- arg.value = json_object_get_string(value);
- } else {
- arg.name = NULL;
- arg.value = NULL;
- }
- arg.path = NULL;
- return arg;
+ return afb_msg_json_get_arg(wsreq_json(wsreq), name);
}
-static void aws_emit(struct afb_ws_json1 *aws, int code, const char *id, size_t idlen, struct json_object *data, const char *token)
+static void wsreq_fail(struct afb_wsreq *wsreq, const char *status, const char *info)
{
- json_object *msg;
- const char *txt;
-
- /* pack the message */
- msg = json_object_new_array();
- json_object_array_add(msg, json_object_new_int(code));
- json_object_array_add(msg, json_object_new_string_len(id, (int)idlen));
- json_object_array_add(msg, data);
- if (token)
- json_object_array_add(msg, json_object_new_string(token));
-
- /* emits the reply */
- txt = json_object_to_json_string(msg);
- afb_ws_text(aws->ws, txt, strlen(txt));
- json_object_put(msg);
+ int rc;
+ rc = afb_wsj1_reply_error_j(wsreq->msgj1, afb_msg_json_reply_error(status, info, &wsreq->context, NULL), afb_context_sent_token(&wsreq->context));
+ if (rc)
+ ERROR("Can't send fail reply: %m");
}
-static void wsreq_reply(struct afb_wsreq *wsreq, int retcode, const char *status, const char *info, json_object *resp)
+static void wsreq_success(struct afb_wsreq *wsreq, json_object *obj, const char *info)
{
- const char *uuid = afb_context_sent_uuid(&wsreq->context);
- const char *token = afb_context_sent_token(&wsreq->context);
- struct json_object *reply = afb_msg_json_reply(status, info, resp, token, uuid);
- aws_emit(wsreq->aws, retcode, wsreq->id, wsreq->idlen, reply, token);
+ int rc;
+ rc = afb_wsj1_reply_ok_j(wsreq->msgj1, afb_msg_json_reply_ok(info, obj, &wsreq->context, NULL), afb_context_sent_token(&wsreq->context));
+ if (rc)
+ ERROR("Can't send success reply: %m");
}
-static void wsreq_fail(struct afb_wsreq *wsreq, const char *status, const char *info)
+static const char *wsreq_raw(struct afb_wsreq *wsreq, size_t *size)
{
- wsreq_reply(wsreq, RETERR, status, info, NULL);
+ const char *result = afb_wsj1_msg_object_s(wsreq->msgj1);
+ if (size != NULL)
+ *size = strlen(result);
+ return result;
}
-static void wsreq_success(struct afb_wsreq *wsreq, json_object *obj, const char *info)
+static void wsreq_send(struct afb_wsreq *wsreq, const char *buffer, size_t size)
{
- wsreq_reply(wsreq, RETOK, "success", info, obj);
+ int rc;
+ rc = afb_wsj1_reply_ok_s(wsreq->msgj1, buffer, afb_context_sent_token(&wsreq->context));
+ if (rc)
+ ERROR("Can't send raw reply: %m");
}
-static const char *wsreq_raw(struct afb_wsreq *wsreq, size_t *size)
+static int wsreq_subscribe(struct afb_wsreq *wsreq, struct afb_event event)
{
- *size = wsreq->objlen;
- return wsreq->obj;
+ return afb_evt_add_watch(wsreq->aws->listener, event);
}
-static void wsreq_send(struct afb_wsreq *wsreq, const char *buffer, size_t size)
+static int wsreq_unsubscribe(struct afb_wsreq *wsreq, struct afb_event event)
{
- afb_ws_text(wsreq->aws->ws, buffer, size);
+ return afb_evt_remove_watch(wsreq->aws->listener, event);
}
-static void aws_send_event(struct afb_ws_json1 *aws, const char *event, struct json_object *object)
+static void wsreq_subcall(struct afb_wsreq *wsreq, const char *api, const char *verb, struct json_object *args, void (*callback)(void*, int, struct json_object*), void *closure)
{
- aws_emit(aws, EVENT, event, strlen(event), afb_msg_json_event(event, object), NULL);
+ afb_subcall(&wsreq->context, api, verb, args, callback, closure, (struct afb_req){ .itf = &afb_ws_json1_req_itf, .closure = wsreq });
}