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 #define WEBSOCKET_CODE_POLICY_VIOLATION  1008
  43 #define WEBSOCKET_CODE_INTERNAL_ERROR    1011
  44
  45 static void wsj1_on_hangup(struct afb_wsj1 *wsj1);
  46 static void wsj1_on_text(struct afb_wsj1 *wsj1, char *text, size_t size);
  47 static struct afb_wsj1_msg *wsj1_msg_make(struct afb_wsj1 *wsj1, char *text, size_t size);
  48
  49 static struct afb_ws_itf wsj1_itf = {
  50         .on_hangup = (void*)wsj1_on_hangup,
  51         .on_text = (void*)wsj1_on_text
  52 };
  53
  54 struct wsj1_call
  55 {
  56         struct wsj1_call *next;
  57         void (*callback)(void *, struct afb_wsj1_msg *);
  58         void *closure;
  59         char id[16];
  60 };
  61
  62 struct afb_wsj1_msg
  63 {
  64         int refcount;
  65         struct afb_wsj1 *wsj1;
  66         struct afb_wsj1_msg *next, *previous;
  67         char *text;
  68         int code;
  69         const char *id;
  70         const char *api;
  71         const char *verb;
  72         const char *event;
  73         const char *object_s;
  74         size_t object_s_length;
  75         const char *token;
  76         struct json_object *object_j;
  77 };
  78
  79 struct afb_wsj1
  80 {
  81         int refcount;
  82         int genid;
  83         struct afb_wsj1_itf *itf;
  84         void *closure;
  85         struct json_tokener *tokener;
  86         struct afb_ws *ws;
  87         struct afb_wsj1_msg *messages;
  88         struct wsj1_call *calls;
  89         pthread_mutex_t mutex;
  90 };
  91
  92 struct afb_wsj1 *afb_wsj1_create(struct fdev *fdev, struct afb_wsj1_itf *itf, void *closure)
  93 {
  94         struct afb_wsj1 *result;
  95
  96         assert(fdev);
  97         assert(itf);
  98         assert(itf->on_call);
  99
 100         result = calloc(1, sizeof * result);
 101         if (result == NULL)
 102                 goto error;
 103
 104         result->refcount = 1;
 105         result->itf = itf;
 106         result->closure = closure;
 107         pthread_mutex_init(&result->mutex, NULL);
 108
 109         result->tokener = json_tokener_new();
 110         if (result->tokener == NULL)
 111                 goto error2;
 112
 113         result->ws = afb_ws_create(fdev, &wsj1_itf, result);
 114         if (result->ws == NULL)
 115                 goto error3;
 116
 117         return result;
 118
 119 error3:
 120         json_tokener_free(result->tokener);
 121 error2:
 122         free(result);
 123 error:
 124         fdev_unref(fdev);
 125         return NULL;
 126 }
 127
 128 void afb_wsj1_addref(struct afb_wsj1 *wsj1)
 129 {
 130         if (wsj1)
 131                 __atomic_add_fetch(&wsj1->refcount, 1, __ATOMIC_RELAXED);
 132 }
 133
 134 void afb_wsj1_unref(struct afb_wsj1 *wsj1)
 135 {
 136         if (wsj1 && !__atomic_sub_fetch(&wsj1->refcount, 1, __ATOMIC_RELAXED)) {
 137                 afb_ws_destroy(wsj1->ws);
 138                 json_tokener_free(wsj1->tokener);
 139                 free(wsj1);
 140         }
 141 }
 142
 143 static void wsj1_on_hangup(struct afb_wsj1 *wsj1)
 144 {
 145         struct wsj1_call *call, *ncall;
 146         struct afb_wsj1_msg *msg;
 147         char *text;
 148         int len;
 149
 150         static const char error_object_str[] = "{"
 151                 "\"jtype\":\"afb-reply\","
 152                 "\"request\":{"
 153                         "\"status\":\"disconnected\","
 154                         "\"info\":\"server hung up\"}}";
 155
 156         ncall = __atomic_exchange_n(&wsj1->calls, NULL, __ATOMIC_RELAXED);
 157         while (ncall) {
 158                 call = ncall;
 159                 ncall = call->next;
 160                 len = asprintf(&text, "[%d,\"%s\",%s]", RETERR, call->id, error_object_str);
 161                 if (len > 0) {
 162                         msg = wsj1_msg_make(wsj1, text, (size_t)len);
 163                         if (msg != NULL) {
 164                                 call->callback(call->closure, msg);
 165                                 afb_wsj1_msg_unref(msg);
 166                         }
 167                 }
 168                 free(call);
 169         }
 170
 171         if (wsj1->itf->on_hangup != NULL)
 172                 wsj1->itf->on_hangup(wsj1->closure, wsj1);
 173 }
 174
 175
 176 static struct wsj1_call *wsj1_locked_call_search(struct afb_wsj1 *wsj1, const char *id, int remove)
 177 {
 178         struct wsj1_call *r, **p;
 179
 180         p = &wsj1->calls;
 181         while((r = *p) != NULL) {
 182                 if (strcmp(r->id, id) == 0) {
 183                         if (remove)
 184                                 *p = r->next;
 185                         break;
 186                 }
 187                 p = &r->next;
 188         }
 189
 190         return r;
 191 }
 192
 193 static struct wsj1_call *wsj1_call_search(struct afb_wsj1 *wsj1, const char *id, int remove)
 194 {
 195         struct wsj1_call *r;
 196
 197         pthread_mutex_lock(&wsj1->mutex);
 198         r = wsj1_locked_call_search(wsj1, id, remove);
 199         pthread_mutex_unlock(&wsj1->mutex);
 200
 201         return r;
 202 }
 203
 204 static struct wsj1_call *wsj1_call_create(struct afb_wsj1 *wsj1, void (*on_reply)(void*,struct afb_wsj1_msg*), void *closure)
 205 {
 206         struct wsj1_call *call = malloc(sizeof *call);
 207         if (call == NULL)
 208                 errno = ENOMEM;
 209         else {
 210                 pthread_mutex_lock(&wsj1->mutex);
 211                 do {
 212                         if (wsj1->genid == 0)
 213                                 wsj1->genid = 999999;
 214                         sprintf(call->id, "%d", wsj1->genid--);
 215                 } while (wsj1_locked_call_search(wsj1, call->id, 0) != NULL);
 216                 call->callback = on_reply;
 217                 call->closure = closure;
 218                 call->next = wsj1->calls;
 219                 wsj1->calls = call;
 220                 pthread_mutex_unlock(&wsj1->mutex);
 221         }
 222         return call;
 223 }
 224
 225
 226 static int wsj1_msg_scan(char *text, size_t items[10][2])
 227 {
 228         char *pos, *beg, *end, c;
 229         int aux, n = 0;
 230
 231         /* scan */
 232         pos = text;
 233
 234         /* scans: [ */
 235         while(*pos == ' ') pos++;
 236         if (*pos++ != '[') goto bad_scan;
 237
 238         /* scans list */
 239         while(*pos == ' ') pos++;
 240         if (*pos != ']') {
 241                 for (;;) {
 242                         if (n == 10)
 243                                 goto bad_scan;
 244                         beg = pos;
 245                         aux = 0;
 246                         while (aux != 0 || (*pos != ',' && *pos != ']')) {
 247                                 switch(*pos++) {
 248                                 case '{': case '[': aux++; break;
 249                                 case '}': case ']': if (aux--) break;
 250                                 case 0: goto bad_scan;
 251                                 case '"':
 252                                         do {
 253                                                 switch(c = *pos++) {
 254                                                 case '\\': if (*pos++) break;
 255                                                 case 0: goto bad_scan;
 256                                                 }
 257                                         } while(c != '"');
 258                                 }
 259                         }
 260                         end = pos;
 261                         while (end > beg && end[-1] == ' ')
 262                                 end--;
 263                         items[n][0] = beg - text; /* start offset */
 264                         items[n][1] = end - beg;  /* length */
 265                         n++;
 266                         if (*pos == ']')
 267                                 break;
 268                         while(*++pos == ' ');
 269                 }
 270         }
 271         while(*++pos == ' ');
 272         if (*pos) goto bad_scan;
 273         return n;
 274
 275 bad_scan:
 276         return -1;
 277 }
 278
 279 static char *wsj1_msg_parse_extract(char *text, size_t offset, size_t size)
 280 {
 281         text[offset + size] = 0;
 282         return text + offset;
 283 }
 284
 285 static char *wsj1_msg_parse_string(char *text, size_t offset, size_t size)
 286 {
 287         if (size > 1 && text[offset] == '"') {
 288                 offset += 1;
 289                 size -= 2;
 290         }
 291         return wsj1_msg_parse_extract(text, offset, size);
 292 }
 293
 294 static struct afb_wsj1_msg *wsj1_msg_make(struct afb_wsj1 *wsj1, char *text, size_t size)
 295 {
 296         size_t items[10][2];
 297         int n;
 298         struct afb_wsj1_msg *msg;
 299         char *verb;
 300
 301         /* allocate */
 302         msg = calloc(1, sizeof *msg);
 303         if (msg == NULL) {
 304                 errno = ENOMEM;
 305                 goto alloc_error;
 306         }
 307
 308         /* scan */
 309         n = wsj1_msg_scan(text, items);
 310         if (n < 0)
 311                 goto bad_header;
 312
 313         /* scans code: 2|3|4|5 */
 314         if (items[0][1] != 1) goto bad_header;
 315         switch (text[items[0][0]]) {
 316         case '2': msg->code = CALL; break;
 317         case '3': msg->code = RETOK; break;
 318         case '4': msg->code = RETERR; break;
 319         case '5': msg->code = EVENT; break;
 320         default: goto bad_header;
 321         }
 322
 323         /* fills the message */
 324         switch (msg->code) {
 325         case CALL:
 326                 if (n != 4 && n != 5) goto bad_header;
 327                 msg->id = wsj1_msg_parse_string(text, items[1][0], items[1][1]);
 328                 msg->api = wsj1_msg_parse_string(text, items[2][0], items[2][1]);
 329                 verb = strchr(msg->api, '/');
 330                 if (verb == NULL) goto bad_header;
 331                 *verb++ = 0;
 332                 if (!*verb || *verb == '/') goto bad_header;
 333                 msg->verb = verb;
 334                 msg->object_s = wsj1_msg_parse_extract(text, items[3][0], items[3][1]);
 335                 msg->object_s_length = items[3][1];
 336                 msg->token = n == 5 ? wsj1_msg_parse_string(text, items[4][0], items[4][1]) : NULL;
 337                 break;
 338         case RETOK:
 339         case RETERR:
 340                 if (n != 3 && n != 4) goto bad_header;
 341                 msg->id = wsj1_msg_parse_string(text, items[1][0], items[1][1]);
 342                 msg->object_s = wsj1_msg_parse_extract(text, items[2][0], items[2][1]);
 343                 msg->object_s_length = items[2][1];
 344                 msg->token = n == 5 ? wsj1_msg_parse_string(text, items[3][0], items[3][1]) : NULL;
 345                 break;
 346         case EVENT:
 347                 if (n != 3) goto bad_header;
 348                 msg->event = wsj1_msg_parse_string(text, items[1][0], items[1][1]);
 349                 msg->object_s = wsj1_msg_parse_extract(text, items[2][0], items[2][1]);
 350                 msg->object_s_length = items[2][1];
 351                 break;
 352         }
 353         /* done */
 354         msg->text = text;
 355
 356         /* fill and record the request */
 357         msg->refcount = 1;
 358         afb_wsj1_addref(wsj1);
 359         msg->wsj1 = wsj1;
 360         pthread_mutex_lock(&wsj1->mutex);
 361         msg->next = wsj1->messages;
 362         if (msg->next != NULL)
 363                 msg->next->previous = msg;
 364         wsj1->messages = msg;
 365         pthread_mutex_unlock(&wsj1->mutex);
 366
 367         return msg;
 368
 369 bad_header:
 370         errno = EBADMSG;
 371         free(msg);
 372
 373 alloc_error:
 374         free(text);
 375         return NULL;
 376 }
 377
 378 static void wsj1_on_text(struct afb_wsj1 *wsj1, char *text, size_t size)
 379 {
 380         struct wsj1_call *call;
 381         struct afb_wsj1_msg *msg;
 382
 383         /* allocate */
 384         msg = wsj1_msg_make(wsj1, text, size);
 385         if (msg == NULL) {
 386                 afb_ws_close(wsj1->ws, errno == EBADMSG
 387                         ? WEBSOCKET_CODE_POLICY_VIOLATION
 388                         : WEBSOCKET_CODE_INTERNAL_ERROR, NULL);
 389                 return;
 390         }
 391
 392         /* handle the message */
 393         switch (msg->code) {
 394         case CALL:
 395                 wsj1->itf->on_call(wsj1->closure, msg->api, msg->verb, msg);
 396                 break;
 397         case RETOK:
 398         case RETERR:
 399                 call = wsj1_call_search(wsj1, msg->id, 1);
 400                 if (call == NULL)
 401                         afb_ws_close(wsj1->ws, WEBSOCKET_CODE_POLICY_VIOLATION, NULL);
 402                 else
 403                         call->callback(call->closure, msg);
 404                 free(call);
 405                 break;
 406         case EVENT:
 407                 if (wsj1->itf->on_event != NULL)
 408                         wsj1->itf->on_event(wsj1->closure, msg->event, msg);
 409                 break;
 410         }
 411         afb_wsj1_msg_unref(msg);
 412 }
 413
 414 void afb_wsj1_msg_addref(struct afb_wsj1_msg *msg)
 415 {
 416         if (msg != NULL)
 417                 __atomic_add_fetch(&msg->refcount, 1, __ATOMIC_RELAXED);
 418 }
 419
 420 void afb_wsj1_msg_unref(struct afb_wsj1_msg *msg)
 421 {
 422         if (msg != NULL && !__atomic_sub_fetch(&msg->refcount, 1, __ATOMIC_RELAXED)) {
 423                 /* unlink the message */
 424                 pthread_mutex_lock(&msg->wsj1->mutex);
 425                 if (msg->next != NULL)
 426                         msg->next->previous = msg->previous;
 427                 if (msg->previous == NULL)
 428                         msg->wsj1->messages = msg->next;
 429                 else
 430                         msg->previous->next = msg->next;
 431                 pthread_mutex_unlock(&msg->wsj1->mutex);
 432                 /* free ressources */
 433                 afb_wsj1_unref(msg->wsj1);
 434                 json_object_put(msg->object_j);
 435                 free(msg->text);
 436                 free(msg);
 437         }
 438 }
 439
 440 const char *afb_wsj1_msg_object_s(struct afb_wsj1_msg *msg)
 441 {
 442         return msg->object_s;
 443 }
 444
 445 struct json_object *afb_wsj1_msg_object_j(struct afb_wsj1_msg *msg)
 446 {
 447         enum json_tokener_error jerr;
 448         struct json_object *object = msg->object_j;
 449         if (object == NULL) {
 450                 pthread_mutex_lock(&msg->wsj1->mutex);
 451                 json_tokener_reset(msg->wsj1->tokener);
 452                 object = json_tokener_parse_ex(msg->wsj1->tokener, msg->object_s, 1 + (int)msg->object_s_length);
 453                 jerr = json_tokener_get_error(msg->wsj1->tokener);
 454                 pthread_mutex_unlock(&msg->wsj1->mutex);
 455                 if (jerr != json_tokener_success) {
 456                         /* lazy error detection of json request. Is it to improve? */
 457                         object = json_object_new_string_len(msg->object_s, (int)msg->object_s_length);
 458                 }
 459                 msg->object_j = object;
 460         }
 461         return object;
 462 }
 463
 464 int afb_wsj1_msg_is_call(struct afb_wsj1_msg *msg)
 465 {
 466         return msg->code == CALL;
 467 }
 468
 469 int afb_wsj1_msg_is_reply(struct afb_wsj1_msg *msg)
 470 {
 471         return msg->code == RETOK || msg->code == RETERR;
 472 }
 473
 474 int afb_wsj1_msg_is_reply_ok(struct afb_wsj1_msg *msg)
 475 {
 476         return msg->code == RETOK;
 477 }
 478
 479 int afb_wsj1_msg_is_reply_error(struct afb_wsj1_msg *msg)
 480 {
 481         return msg->code == RETERR;
 482 }
 483
 484 int afb_wsj1_msg_is_event(struct afb_wsj1_msg *msg)
 485 {
 486         return msg->code == EVENT;
 487 }
 488
 489 const char *afb_wsj1_msg_api(struct afb_wsj1_msg *msg)
 490 {
 491         return msg->api;
 492 }
 493
 494 const char *afb_wsj1_msg_verb(struct afb_wsj1_msg *msg)
 495 {
 496         return msg->verb;
 497 }
 498
 499 const char *afb_wsj1_msg_event(struct afb_wsj1_msg *msg)
 500 {
 501         return msg->event;
 502 }
 503
 504 const char *afb_wsj1_msg_token(struct afb_wsj1_msg *msg)
 505 {
 506         return msg->token;
 507 }
 508
 509 struct afb_wsj1 *afb_wsj1_msg_wsj1(struct afb_wsj1_msg *msg)
 510 {
 511         return msg->wsj1;
 512 }
 513
 514 int afb_wsj1_close(struct afb_wsj1 *wsj1, uint16_t code, const char *text)
 515 {
 516         return afb_ws_close(wsj1->ws, code, text);
 517 }
 518
 519 static int wsj1_send_isot(struct afb_wsj1 *wsj1, int i1, const char *s1, const char *o1, const char *t1)
 520 {
 521         char code[2] = { (char)('0' + i1), 0 };
 522         return afb_ws_texts(wsj1->ws, "[", code, ",\"", s1, "\",", o1 == NULL ? "null" : o1, t1 != NULL ? ",\"" : "]", t1, "\"]", NULL);
 523 }
 524
 525 static int wsj1_send_issot(struct afb_wsj1 *wsj1, int i1, const char *s1, const char *s2, const char *o1, const char *t1)
 526 {
 527         char code[2] = { (char)('0' + i1), 0 };
 528         return afb_ws_texts(wsj1->ws, "[", code, ",\"", s1, "\",\"", s2, "\",", o1 == NULL ? "null" : o1, t1 != NULL ? ",\"" : "]", t1, "\"]", NULL);
 529 }
 530
 531 int afb_wsj1_send_event_j(struct afb_wsj1 *wsj1, const char *event, struct json_object *object)
 532 {
 533         const char *objstr = json_object_to_json_string_ext(object, JSON_C_TO_STRING_PLAIN|JSON_C_TO_STRING_NOSLASHESCAPE);
 534         int rc = afb_wsj1_send_event_s(wsj1, event, objstr);
 535         json_object_put(object);
 536         return rc;
 537 }
 538
 539 int afb_wsj1_send_event_s(struct afb_wsj1 *wsj1, const char *event, const char *object)
 540 {
 541         return wsj1_send_isot(wsj1, EVENT, event, object, NULL);
 542 }
 543
 544 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)
 545 {
 546         const char *objstr = json_object_to_json_string_ext(object, JSON_C_TO_STRING_PLAIN|JSON_C_TO_STRING_NOSLASHESCAPE);
 547         int rc = afb_wsj1_call_s(wsj1, api, verb, objstr, on_reply, closure);
 548         json_object_put(object);
 549         return rc;
 550 }
 551
 552 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)
 553 {
 554         int rc;
 555         struct wsj1_call *call;
 556         char *tag;
 557
 558         /* allocates the call */
 559         call = wsj1_call_create(wsj1, on_reply, closure);
 560         if (call == NULL) {
 561                 errno = ENOMEM;
 562                 return -1;
 563         }
 564
 565         /* makes the tag */
 566         tag = alloca(2 + strlen(api) + strlen(verb));
 567         stpcpy(stpcpy(stpcpy(tag, api), "/"), verb);
 568
 569         /* makes the call */
 570         rc = wsj1_send_issot(wsj1, CALL, call->id, tag, object, NULL);
 571         if (rc < 0) {
 572                 wsj1_call_search(wsj1, call->id, 1);
 573                 free(call);
 574         }
 575         return rc;
 576 }
 577
 578 int afb_wsj1_reply_j(struct afb_wsj1_msg *msg, struct json_object *object, const char *token, int iserror)
 579 {
 580         const char *objstr = json_object_to_json_string_ext(object, JSON_C_TO_STRING_PLAIN|JSON_C_TO_STRING_NOSLASHESCAPE);
 581         int rc = afb_wsj1_reply_s(msg, objstr, token, iserror);
 582         json_object_put(object);
 583         return rc;
 584 }
 585
 586 int afb_wsj1_reply_s(struct afb_wsj1_msg *msg, const char *object, const char *token, int iserror)
 587 {
 588         return wsj1_send_isot(msg->wsj1, iserror ? RETERR : RETOK, msg->id, object, token);
 589 }
 590