src/afb-wsj1.c

   1 /*
   2  * Copyright (C) 2016, 2017, 2018 "IoT.bzh"
   3  * Author: José Bollo <jose.bollo@iot.bzh>
   4  *
   5  * Licensed under the Apache License, Version 2.0 (the "License");
   6  * you may not use this file except in compliance with the License.
   7  * You may obtain a copy of the License at
   8  *
   9  *   http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  * Unless required by applicable law or agreed to in writing, software
  12  * distributed under the License is distributed on an "AS IS" BASIS,
  13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14  * See the License for the specific language governing permissions and
  15  * limitations under the License.
  16  */
  17
  18 #define _GNU_SOURCE
  19
  20 #include <stdlib.h>
  21 #include <unistd.h>
  22 #include <assert.h>
  23 #include <errno.h>
  24 #include <string.h>
  25 #include <stdio.h>
  26 #include <pthread.h>
  27
  28 #include <json-c/json.h>
  29 #if !defined(JSON_C_TO_STRING_NOSLASHESCAPE)
  30 #define JSON_C_TO_STRING_NOSLASHESCAPE 0
  31 #endif
  32
  33 #include "afb-ws.h"
  34 #include "afb-wsj1.h"
  35 #include "fdev.h"
  36
  37 #define CALL 2
  38 #define RETOK 3
  39 #define RETERR 4
  40 #define EVENT 5
  41
  42 static void wsj1_on_hangup(struct afb_wsj1 *wsj1);
  43 static void wsj1_on_text(struct afb_wsj1 *wsj1, char *text, size_t size);
  44
  45 static struct afb_ws_itf wsj1_itf = {
  46         .on_hangup = (void*)wsj1_on_hangup,
  47         .on_text = (void*)wsj1_on_text
  48 };
  49
  50 struct wsj1_call
  51 {
  52         struct wsj1_call *next;
  53         void (*callback)(void *, struct afb_wsj1_msg *);
  54         void *closure;
  55         char id[16];
  56 };
  57
  58 struct afb_wsj1_msg
  59 {
  60         int refcount;
  61         struct afb_wsj1 *wsj1;
  62         struct afb_wsj1_msg *next, *previous;
  63         char *text;
  64         int code;
  65         char *id;
  66         char *api;
  67         char *verb;
  68         char *event;
  69         char *object_s;
  70         size_t object_s_length;
  71         char *token;
  72         struct json_object *object_j;
  73 };
  74
  75 struct afb_wsj1
  76 {
  77         int refcount;
  78         int genid;
  79         struct afb_wsj1_itf *itf;
  80         void *closure;
  81         struct json_tokener *tokener;
  82         struct afb_ws *ws;
  83         struct afb_wsj1_msg *messages;
  84         struct wsj1_call *calls;
  85         pthread_mutex_t mutex;
  86 };
  87
  88 struct afb_wsj1 *afb_wsj1_create(struct fdev *fdev, struct afb_wsj1_itf *itf, void *closure)
  89 {
  90         struct afb_wsj1 *result;
  91
  92         assert(fdev);
  93         assert(itf);
  94         assert(itf->on_call);
  95
  96         result = calloc(1, sizeof * result);
  97         if (result == NULL)
  98                 goto error;
  99
 100         result->refcount = 1;
 101         result->itf = itf;
 102         result->closure = closure;
 103         pthread_mutex_init(&result->mutex, NULL);
 104
 105         result->tokener = json_tokener_new();
 106         if (result->tokener == NULL)
 107                 goto error2;
 108
 109         result->ws = afb_ws_create(fdev, &wsj1_itf, result);
 110         if (result->ws == NULL)
 111                 goto error3;
 112
 113         return result;
 114
 115 error3:
 116         json_tokener_free(result->tokener);
 117 error2:
 118         free(result);
 119 error:
 120         fdev_unref(fdev);
 121         return NULL;
 122 }
 123
 124 void afb_wsj1_addref(struct afb_wsj1 *wsj1)
 125 {
 126         if (wsj1)
 127                 __atomic_add_fetch(&wsj1->refcount, 1, __ATOMIC_RELAXED);
 128 }
 129
 130 void afb_wsj1_unref(struct afb_wsj1 *wsj1)
 131 {
 132         if (wsj1 && !__atomic_sub_fetch(&wsj1->refcount, 1, __ATOMIC_RELAXED)) {
 133                 afb_ws_destroy(wsj1->ws);
 134                 json_tokener_free(wsj1->tokener);
 135                 free(wsj1);
 136         }
 137 }
 138
 139 static void wsj1_on_hangup(struct afb_wsj1 *wsj1)
 140 {
 141         if (wsj1->itf->on_hangup != NULL)
 142                 wsj1->itf->on_hangup(wsj1->closure, wsj1);
 143 }
 144
 145
 146 static struct wsj1_call *wsj1_locked_call_search(struct afb_wsj1 *wsj1, const char *id, int remove)
 147 {
 148         struct wsj1_call *r, **p;
 149
 150         p = &wsj1->calls;
 151         while((r = *p) != NULL) {
 152                 if (strcmp(r->id, id) == 0) {
 153                         if (remove)
 154                                 *p = r->next;
 155                         break;
 156                 }
 157                 p = &r->next;
 158         }
 159
 160         return r;
 161 }
 162
 163 static struct wsj1_call *wsj1_call_search(struct afb_wsj1 *wsj1, const char *id, int remove)
 164 {
 165         struct wsj1_call *r;
 166
 167         pthread_mutex_lock(&wsj1->mutex);
 168         r = wsj1_locked_call_search(wsj1, id, remove);
 169         pthread_mutex_unlock(&wsj1->mutex);
 170
 171         return r;
 172 }
 173
 174 static struct wsj1_call *wsj1_call_create(struct afb_wsj1 *wsj1, void (*on_reply)(void*,struct afb_wsj1_msg*), void *closure)
 175 {
 176         struct wsj1_call *call = malloc(sizeof *call);
 177         if (call == NULL)
 178                 errno = ENOMEM;
 179         else {
 180                 pthread_mutex_lock(&wsj1->mutex);
 181                 do {
 182                         if (wsj1->genid == 0)
 183                                 wsj1->genid = 999999;
 184                         sprintf(call->id, "%d", wsj1->genid--);
 185                 } while (wsj1_locked_call_search(wsj1, call->id, 0) != NULL);
 186                 call->callback = on_reply;
 187                 call->closure = closure;
 188                 call->next = wsj1->calls;
 189                 wsj1->calls = call;
 190                 pthread_mutex_unlock(&wsj1->mutex);
 191         }
 192         return call;
 193 }
 194
 195
 196 static int wsj1_msg_scan(char *text, size_t items[10][2])
 197 {
 198         char *pos, *beg, *end, c;
 199         int aux, n = 0;
 200
 201         /* scan */
 202         pos = text;
 203
 204         /* scans: [ */
 205         while(*pos == ' ') pos++;
 206         if (*pos++ != '[') goto bad_scan;
 207
 208         /* scans list */
 209         while(*pos == ' ') pos++;
 210         if (*pos != ']') {
 211                 for (;;) {
 212                         if (n == 10)
 213                                 goto bad_scan;
 214                         beg = pos;
 215                         aux = 0;
 216                         while (aux != 0 || (*pos != ',' && *pos != ']')) {
 217                                 switch(*pos++) {
 218                                 case '{': case '[': aux++; break;
 219                                 case '}': case ']': if (aux--) break;
 220                                 case 0: goto bad_scan;
 221                                 case '"':
 222                                         do {
 223                                                 switch(c = *pos++) {
 224                                                 case '\\': if (*pos++) break;
 225                                                 case 0: goto bad_scan;
 226                                                 }
 227                                         } while(c != '"');
 228                                 }
 229                         }
 230                         end = pos;
 231                         while (end > beg && end[-1] == ' ')
 232                                 end--;
 233                         items[n][0] = beg - text; /* start offset */
 234                         items[n][1] = end - beg;  /* length */
 235                         n++;
 236                         if (*pos == ']')
 237                                 break;
 238                         while(*++pos == ' ');
 239                 }
 240         }
 241         while(*++pos == ' ');
 242         if (*pos) goto bad_scan;
 243         return n;
 244
 245 bad_scan:
 246         return -1;
 247 }
 248
 249 static char *wsj1_msg_parse_extract(char *text, size_t offset, size_t size)
 250 {
 251         text[offset + size] = 0;
 252         return text + offset;
 253 }
 254
 255 static char *wsj1_msg_parse_string(char *text, size_t offset, size_t size)
 256 {
 257         if (size > 1 && text[offset] == '"') {
 258                 offset += 1;
 259                 size -= 2;
 260         }
 261         return wsj1_msg_parse_extract(text, offset, size);
 262 }
 263
 264 static void wsj1_on_text(struct afb_wsj1 *wsj1, char *text, size_t size)
 265 {
 266         size_t items[10][2];
 267         int n;
 268         struct afb_wsj1_msg *msg;
 269         struct wsj1_call *call = NULL;
 270
 271         /* allocate */
 272         msg = calloc(1, sizeof *msg);
 273         if (msg == NULL)
 274                 goto alloc_error;
 275
 276         /* scan */
 277         n = wsj1_msg_scan(text, items);
 278         if (n < 0)
 279                 goto bad_header;
 280
 281         /* scans code: 2|3|4|5 */
 282         if (items[0][1] != 1) goto bad_header;
 283         switch (text[items[0][0]]) {
 284         case '2': msg->code = CALL; break;
 285         case '3': msg->code = RETOK; break;
 286         case '4': msg->code = RETERR; break;
 287         case '5': msg->code = EVENT; break;
 288         default: goto bad_header;
 289         }
 290
 291         /* fills the message */
 292         switch (msg->code) {
 293         case CALL:
 294                 if (n != 4 && n != 5) goto bad_header;
 295                 msg->id = wsj1_msg_parse_string(text, items[1][0], items[1][1]);
 296                 msg->api = wsj1_msg_parse_string(text, items[2][0], items[2][1]);
 297                 msg->verb = strchr(msg->api, '/');
 298                 if (msg->verb == NULL) goto bad_header;
 299                 *msg->verb++ = 0;
 300                 msg->object_s = wsj1_msg_parse_extract(text, items[3][0], items[3][1]);
 301                 msg->object_s_length = items[3][1];
 302                 msg->token = n == 5 ? wsj1_msg_parse_string(text, items[4][0], items[4][1]) : NULL;
 303                 break;
 304         case RETOK:
 305         case RETERR:
 306                 if (n != 3 && n != 4) goto bad_header;
 307                 msg->id = wsj1_msg_parse_string(text, items[1][0], items[1][1]);
 308                 call = wsj1_call_search(wsj1, msg->id, 1);
 309                 if (call == NULL) goto bad_header;
 310                 msg->object_s = wsj1_msg_parse_extract(text, items[2][0], items[2][1]);
 311                 msg->object_s_length = items[2][1];
 312                 msg->token = n == 5 ? wsj1_msg_parse_string(text, items[3][0], items[3][1]) : NULL;
 313                 break;
 314         case EVENT:
 315                 if (n != 3) goto bad_header;
 316                 msg->event = wsj1_msg_parse_string(text, items[1][0], items[1][1]);
 317                 msg->object_s = wsj1_msg_parse_extract(text, items[2][0], items[2][1]);
 318                 msg->object_s_length = items[2][1];
 319                 break;
 320         }
 321         /* done */
 322         msg->text = text;
 323
 324         /* fill and record the request */
 325         msg->refcount = 1;
 326         afb_wsj1_addref(wsj1);
 327         msg->wsj1 = wsj1;
 328         pthread_mutex_lock(&wsj1->mutex);
 329         msg->next = wsj1->messages;
 330         if (msg->next != NULL)
 331                 msg->next->previous = msg;
 332         wsj1->messages = msg;
 333         pthread_mutex_unlock(&wsj1->mutex);
 334
 335         /* incoke the handler */
 336         switch (msg->code) {
 337         case CALL:
 338                 wsj1->itf->on_call(wsj1->closure, msg->api, msg->verb, msg);
 339                 break;
 340         case RETOK:
 341         case RETERR:
 342                 call->callback(call->closure, msg);
 343                 free(call);
 344                 break;
 345         case EVENT:
 346                 if (wsj1->itf->on_event != NULL)
 347                         wsj1->itf->on_event(wsj1->closure, msg->event, msg);
 348                 break;
 349         }
 350         afb_wsj1_msg_unref(msg);
 351         return;
 352
 353 bad_header:
 354         free(msg);
 355 alloc_error:
 356         free(text);
 357         afb_ws_close(wsj1->ws, 1008, NULL);
 358 }
 359
 360 void afb_wsj1_msg_addref(struct afb_wsj1_msg *msg)
 361 {
 362         if (msg != NULL)
 363                 __atomic_add_fetch(&msg->refcount, 1, __ATOMIC_RELAXED);
 364 }
 365
 366 void afb_wsj1_msg_unref(struct afb_wsj1_msg *msg)
 367 {
 368         if (msg != NULL && !__atomic_sub_fetch(&msg->refcount, 1, __ATOMIC_RELAXED)) {
 369                 /* unlink the message */
 370                 pthread_mutex_lock(&msg->wsj1->mutex);
 371                 if (msg->next != NULL)
 372                         msg->next->previous = msg->previous;
 373                 if (msg->previous == NULL)
 374                         msg->wsj1->messages = msg->next;
 375                 else
 376                         msg->previous->next = msg->next;
 377                 pthread_mutex_unlock(&msg->wsj1->mutex);
 378                 /* free ressources */
 379                 afb_wsj1_unref(msg->wsj1);
 380                 json_object_put(msg->object_j);
 381                 free(msg->text);
 382                 free(msg);
 383         }
 384 }
 385
 386 const char *afb_wsj1_msg_object_s(struct afb_wsj1_msg *msg)
 387 {
 388         return msg->object_s;
 389 }
 390
 391 struct json_object *afb_wsj1_msg_object_j(struct afb_wsj1_msg *msg)
 392 {
 393         enum json_tokener_error jerr;
 394         struct json_object *object = msg->object_j;
 395         if (object == NULL) {
 396                 pthread_mutex_lock(&msg->wsj1->mutex);
 397                 json_tokener_reset(msg->wsj1->tokener);
 398                 object = json_tokener_parse_ex(msg->wsj1->tokener, msg->object_s, 1 + (int)msg->object_s_length);
 399                 jerr = json_tokener_get_error(msg->wsj1->tokener);
 400                 pthread_mutex_unlock(&msg->wsj1->mutex);
 401                 if (jerr != json_tokener_success) {
 402                         /* lazy error detection of json request. Is it to improve? */
 403                         object = json_object_new_string_len(msg->object_s, (int)msg->object_s_length);
 404                 }
 405                 msg->object_j = object;
 406         }
 407         return object;
 408 }
 409
 410 int afb_wsj1_msg_is_call(struct afb_wsj1_msg *msg)
 411 {
 412         return msg->code == CALL;
 413 }
 414
 415 int afb_wsj1_msg_is_reply(struct afb_wsj1_msg *msg)
 416 {
 417         return msg->code == RETOK || msg->code == RETERR;
 418 }
 419
 420 int afb_wsj1_msg_is_reply_ok(struct afb_wsj1_msg *msg)
 421 {
 422         return msg->code == RETOK;
 423 }
 424
 425 int afb_wsj1_msg_is_reply_error(struct afb_wsj1_msg *msg)
 426 {
 427         return msg->code == RETERR;
 428 }
 429
 430 int afb_wsj1_msg_is_event(struct afb_wsj1_msg *msg)
 431 {
 432         return msg->code == EVENT;
 433 }
 434
 435 const char *afb_wsj1_msg_api(struct afb_wsj1_msg *msg)
 436 {
 437         return msg->api;
 438 }
 439
 440 const char *afb_wsj1_msg_verb(struct afb_wsj1_msg *msg)
 441 {
 442         return msg->verb;
 443 }
 444
 445 const char *afb_wsj1_msg_event(struct afb_wsj1_msg *msg)
 446 {
 447         return msg->event;
 448 }
 449
 450 const char *afb_wsj1_msg_token(struct afb_wsj1_msg *msg)
 451 {
 452         return msg->token;
 453 }
 454
 455 struct afb_wsj1 *afb_wsj1_msg_wsj1(struct afb_wsj1_msg *msg)
 456 {
 457         return msg->wsj1;
 458 }
 459
 460 int afb_wsj1_close(struct afb_wsj1 *wsj1, uint16_t code, const char *text)
 461 {
 462         return afb_ws_close(wsj1->ws, code, text);
 463 }
 464
 465 static int wsj1_send_isot(struct afb_wsj1 *wsj1, int i1, const char *s1, const char *o1, const char *t1)
 466 {
 467         char code[2] = { (char)('0' + i1), 0 };
 468         return afb_ws_texts(wsj1->ws, "[", code, ",\"", s1, "\",", o1 == NULL ? "null" : o1, t1 != NULL ? ",\"" : "]", t1, "\"]", NULL);
 469 }
 470
 471 static int wsj1_send_issot(struct afb_wsj1 *wsj1, int i1, const char *s1, const char *s2, const char *o1, const char *t1)
 472 {
 473         char code[2] = { (char)('0' + i1), 0 };
 474         return afb_ws_texts(wsj1->ws, "[", code, ",\"", s1, "\",\"", s2, "\",", o1 == NULL ? "null" : o1, t1 != NULL ? ",\"" : "]", t1, "\"]", NULL);
 475 }
 476
 477 int afb_wsj1_send_event_j(struct afb_wsj1 *wsj1, const char *event, struct json_object *object)
 478 {
 479         const char *objstr = json_object_to_json_string_ext(object, JSON_C_TO_STRING_PLAIN|JSON_C_TO_STRING_NOSLASHESCAPE);
 480         int rc = afb_wsj1_send_event_s(wsj1, event, objstr);
 481         json_object_put(object);
 482         return rc;
 483 }
 484
 485 int afb_wsj1_send_event_s(struct afb_wsj1 *wsj1, const char *event, const char *object)
 486 {
 487         return wsj1_send_isot(wsj1, EVENT, event, object, NULL);
 488 }
 489
 490 int afb_wsj1_call_j(struct afb_wsj1 *wsj1, const char *api, const char *verb, struct json_object *object, void (*on_reply)(void *closure, struct afb_wsj1_msg *msg), void *closure)
 491 {
 492         const char *objstr = json_object_to_json_string_ext(object, JSON_C_TO_STRING_PLAIN|JSON_C_TO_STRING_NOSLASHESCAPE);
 493         int rc = afb_wsj1_call_s(wsj1, api, verb, objstr, on_reply, closure);
 494         json_object_put(object);
 495         return rc;
 496 }
 497
 498 int afb_wsj1_call_s(struct afb_wsj1 *wsj1, const char *api, const char *verb, const char *object, void (*on_reply)(void *closure, struct afb_wsj1_msg *msg), void *closure)
 499 {
 500         int rc;
 501         struct wsj1_call *call;
 502         char *tag;
 503
 504         /* allocates the call */
 505         call = wsj1_call_create(wsj1, on_reply, closure);
 506         if (call == NULL) {
 507                 errno = ENOMEM;
 508                 return -1;
 509         }
 510
 511         /* makes the tag */
 512         tag = alloca(2 + strlen(api) + strlen(verb));
 513         stpcpy(stpcpy(stpcpy(tag, api), "/"), verb);
 514
 515         /* makes the call */
 516         rc = wsj1_send_issot(wsj1, CALL, call->id, tag, object, NULL);
 517         if (rc < 0) {
 518                 wsj1_call_search(wsj1, call->id, 1);
 519                 free(call);
 520         }
 521         return rc;
 522 }
 523
 524 int afb_wsj1_reply_j(struct afb_wsj1_msg *msg, struct json_object *object, const char *token, int iserror)
 525 {
 526         const char *objstr = json_object_to_json_string_ext(object, JSON_C_TO_STRING_PLAIN|JSON_C_TO_STRING_NOSLASHESCAPE);
 527         int rc = afb_wsj1_reply_s(msg, objstr, token, iserror);
 528         json_object_put(object);
 529         return rc;
 530 }
 531
 532 int afb_wsj1_reply_s(struct afb_wsj1_msg *msg, const char *object, const char *token, int iserror)
 533 {
 534         return wsj1_send_isot(msg->wsj1, iserror ? RETERR : RETOK, msg->id, object, token);
 535 }
 536