struct api_ws_event;
struct api_ws_client;
-
+#define CHAR_FOR_CALL 'C'
+#define CHAR_FOR_ANSWER_SUCCESS 'T'
+#define CHAR_FOR_ANSWER_FAIL 'F'
+#define CHAR_FOR_EVT_BROADCAST '*'
+#define CHAR_FOR_EVT_ADD '+'
+#define CHAR_FOR_EVT_DEL '-'
+#define CHAR_FOR_EVT_PUSH '!'
+#define CHAR_FOR_EVT_SUBSCRIBE 'S'
+#define CHAR_FOR_EVT_UNSUBSCRIBE 'U'
/*
*/
struct api_ws_server_req {
struct afb_xreq xreq; /* the xreq */
struct api_ws_client *client; /* the client of the request */
- char *rcvdata; /* the received data to free */
const char *request; /* the readen request as string */
size_t lenreq; /* the length of the request */
uint32_t msgid; /* the incoming request msgid */
};
-static void api_ws_server_req_success_cb(void *closure, struct json_object *obj, const char *info);
-static void api_ws_server_req_fail_cb(void *closure, const char *status, const char *info);
-static void api_ws_server_req_destroy_cb(void *closure);
-static int api_ws_server_req_subscribe_cb(void *closure, struct afb_event event);
-static int api_ws_server_req_unsubscribe_cb(void *closure, struct afb_event event);
+static void api_ws_server_req_success_cb(struct afb_xreq *xreq, struct json_object *obj, const char *info);
+static void api_ws_server_req_fail_cb(struct afb_xreq *xreq, const char *status, const char *info);
+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);
const struct afb_xreq_query_itf afb_api_ws_xreq_itf = {
.success = api_ws_server_req_success_cb,
if (size > 0) {
struct readbuf rb = { .head = data, .end = data + size };
switch (*rb.head++) {
- case 'T': /* success */
+ case CHAR_FOR_ANSWER_SUCCESS: /* success */
api_ws_client_reply_success(closure, &rb);
break;
- case 'F': /* fail */
+ case CHAR_FOR_ANSWER_FAIL: /* fail */
api_ws_client_reply_fail(closure, &rb);
break;
- case '*': /* broadcast */
+ case CHAR_FOR_EVT_BROADCAST: /* broadcast */
api_ws_client_event_broadcast(closure, &rb);
break;
- case '+': /* creates the event */
+ case CHAR_FOR_EVT_ADD: /* creates the event */
api_ws_client_event_create(closure, &rb);
break;
- case '-': /* drops the event */
+ case CHAR_FOR_EVT_DEL: /* drops the event */
api_ws_client_event_drop(closure, &rb);
break;
- case '!': /* pushs the event */
+ case CHAR_FOR_EVT_PUSH: /* pushs the event */
api_ws_client_event_push(closure, &rb);
break;
- case 'S': /* subscribe event for a request */
+ case CHAR_FOR_EVT_SUBSCRIBE: /* subscribe event for a request */
api_ws_client_event_subscribe(closure, &rb);
break;
- case 'U': /* unsubscribe event for a request */
+ case CHAR_FOR_EVT_UNSUBSCRIBE: /* unsubscribe event for a request */
api_ws_client_event_unsubscribe(closure, &rb);
break;
default: /* unexpected message */
raw = afb_xreq_raw(xreq, &szraw);
if (raw == NULL)
goto internal_error;
- if (!api_ws_write_uint32(&wb, memo->msgid)
+ if (!api_ws_write_char(&wb, CHAR_FOR_CALL)
+ || !api_ws_write_uint32(&wb, memo->msgid)
|| !api_ws_write_uint32(&wb, (uint32_t)xreq->context.flags)
|| !api_ws_write_string(&wb, xreq->verb)
|| !api_ws_write_string(&wb, afb_session_uuid(xreq->context.session))
}
/* on call, propagate it to the ws service */
-static void api_ws_server_called(struct api_ws_client *client, struct readbuf *rb, char *data, size_t size)
+static void api_ws_server_on_call(struct api_ws_client *client, struct readbuf *rb)
{
struct api_ws_server_req *wreq;
const char *uuid, *verb;
goto out_of_memory;
wreq->client = client;
- wreq->rcvdata = data;
/* reads the call message data */
if (!api_ws_read_uint32(rb, &wreq->msgid)
|| !api_ws_read_string(rb, &wreq->request, &wreq->lenreq))
goto overflow;
+ 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->xreq.context.flags = flags;
/* makes the call */
- wreq->xreq.refcount = 1;
+ wreq->xreq.cred = afb_cred_addref(client->cred);
wreq->xreq.api = client->api;
wreq->xreq.verb = verb;
- wreq->xreq.query = wreq;
- wreq->xreq.queryitf = &afb_api_ws_xreq_itf;
afb_apis_call(&wreq->xreq);
afb_xreq_unref(&wreq->xreq);
return;
out_of_memory:
overflow:
free(wreq);
- free(data);
api_ws_server_client_unref(client);
}
/* callback when receiving binary data */
static void api_ws_server_on_binary(void *closure, char *data, size_t size)
{
- struct readbuf rb = { .head = data, .end = data + size };
- api_ws_server_called(closure, &rb, data, size);
+ if (size > 0) {
+ struct readbuf rb = { .head = data, .end = data + size };
+ switch (*rb.head++) {
+ case CHAR_FOR_CALL:
+ api_ws_server_on_call(closure, &rb);
+ break;
+ default: /* unexpected message */
+ /* TODO: close the connection */
+ break;
+ }
+ }
+ free(data);
}
/* callback when receiving a hangup */
static void api_ws_server_event_add(void *closure, const char *event, int eventid)
{
- api_ws_server_event_send(closure, '+', event, eventid, NULL);
+ api_ws_server_event_send(closure, CHAR_FOR_EVT_ADD, event, eventid, NULL);
}
static void api_ws_server_event_remove(void *closure, const char *event, int eventid)
{
- api_ws_server_event_send(closure, '-', event, eventid, NULL);
+ api_ws_server_event_send(closure, CHAR_FOR_EVT_DEL, event, eventid, NULL);
}
static void api_ws_server_event_push(void *closure, const char *event, int eventid, struct json_object *object)
{
const char *data = json_object_to_json_string_ext(object, JSON_C_TO_STRING_PLAIN);
- api_ws_server_event_send(closure, '!', event, eventid, data ? : "null");
+ api_ws_server_event_send(closure, CHAR_FOR_EVT_PUSH, event, eventid, data ? : "null");
json_object_put(object);
}
struct writebuf wb = { .count = 0 };
- if (api_ws_write_char(&wb, '*') && api_ws_write_string(&wb, event) && api_ws_write_object(&wb, object)) {
+ if (api_ws_write_char(&wb, CHAR_FOR_EVT_BROADCAST) && api_ws_write_string(&wb, event) && api_ws_write_object(&wb, object)) {
rc = afb_ws_binary_v(client->ws, wb.iovec, wb.count);
if (rc < 0)
ERROR("error while broadcasting event %s", event);
/******************* ws request part for server *****************/
/* decrement the reference count of the request and free/release it on falling to null */
-static void api_ws_server_req_destroy_cb(void *closure)
+static void api_ws_server_req_destroy_cb(struct afb_xreq *xreq)
{
- struct api_ws_server_req *wreq = closure;
+ struct api_ws_server_req *wreq = CONTAINER_OF_XREQ(struct api_ws_server_req, xreq);
afb_context_disconnect(&wreq->xreq.context);
+ afb_cred_unref(wreq->xreq.cred);
json_object_put(wreq->xreq.json);
- free(wreq->rcvdata);
api_ws_server_client_unref(wreq->client);
free(wreq);
}
-static void api_ws_server_req_success_cb(void *closure, struct json_object *obj, const char *info)
+static void api_ws_server_req_success_cb(struct afb_xreq *xreq, struct json_object *obj, const char *info)
{
int rc;
struct writebuf wb = { .count = 0 };
- struct api_ws_server_req *wreq = closure;
+ struct api_ws_server_req *wreq = CONTAINER_OF_XREQ(struct api_ws_server_req, xreq);
- if (api_ws_write_char(&wb, 'T')
+ if (api_ws_write_char(&wb, CHAR_FOR_ANSWER_SUCCESS)
&& api_ws_write_uint32(&wb, wreq->msgid)
&& api_ws_write_uint32(&wb, (uint32_t)wreq->xreq.context.flags)
&& api_ws_write_string(&wb, info ? : "")
json_object_put(obj);
}
-static void api_ws_server_req_fail_cb(void *closure, const char *status, const char *info)
+static void api_ws_server_req_fail_cb(struct afb_xreq *xreq, const char *status, const char *info)
{
int rc;
struct writebuf wb = { .count = 0 };
- struct api_ws_server_req *wreq = closure;
+ struct api_ws_server_req *wreq = CONTAINER_OF_XREQ(struct api_ws_server_req, xreq);
- if (api_ws_write_char(&wb, 'F')
+ if (api_ws_write_char(&wb, CHAR_FOR_ANSWER_FAIL)
&& api_ws_write_uint32(&wb, wreq->msgid)
&& api_ws_write_uint32(&wb, (uint32_t)wreq->xreq.context.flags)
&& api_ws_write_string(&wb, status)
ERROR("error while sending fail");
}
-static int api_ws_server_req_subscribe_cb(void *closure, struct afb_event event)
+static int api_ws_server_req_subscribe_cb(struct afb_xreq *xreq, struct afb_event event)
{
int rc, rc2;
struct writebuf wb = { .count = 0 };
- struct api_ws_server_req *wreq = closure;
+ struct api_ws_server_req *wreq = CONTAINER_OF_XREQ(struct api_ws_server_req, xreq);
rc = afb_evt_add_watch(wreq->client->listener, event);
if (rc < 0)
return rc;
- if (api_ws_write_char(&wb, 'S')
+ if (api_ws_write_char(&wb, CHAR_FOR_EVT_SUBSCRIBE)
&& api_ws_write_uint32(&wb, wreq->msgid)
&& api_ws_write_uint32(&wb, (uint32_t)afb_evt_event_id(event))
&& api_ws_write_string(&wb, afb_evt_event_name(event))) {
return rc;
}
-static int api_ws_server_req_unsubscribe_cb(void *closure, struct afb_event event)
+static int api_ws_server_req_unsubscribe_cb(struct afb_xreq *xreq, struct afb_event event)
{
int rc, rc2;
struct writebuf wb = { .count = 0 };
- struct api_ws_server_req *wreq = closure;
+ struct api_ws_server_req *wreq = CONTAINER_OF_XREQ(struct api_ws_server_req, xreq);
- if (api_ws_write_char(&wb, 'U')
+ if (api_ws_write_char(&wb, CHAR_FOR_EVT_UNSUBSCRIBE)
&& api_ws_write_uint32(&wb, wreq->msgid)
&& api_ws_write_uint32(&wb, (uint32_t)afb_evt_event_id(event))
&& api_ws_write_string(&wb, afb_evt_event_name(event))) {
Code Review / src / app-framework-binder.git / blobdiff