src/afb-api-ws.c

   1 /*
   2  * Copyright (C) 2015, 2016 "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 #define NO_PLUGIN_VERBOSE_MACRO
  20
  21 #include <stdlib.h>
  22 #include <string.h>
  23 #include <assert.h>
  24 #include <fcntl.h>
  25 #include <unistd.h>
  26 #include <errno.h>
  27 #include <endian.h>
  28 #include <netdb.h>
  29 #include <sys/types.h>
  30 #include <sys/socket.h>
  31 #include <sys/un.h>
  32
  33 #include <json-c/json.h>
  34 #include <systemd/sd-event.h>
  35
  36 #include <afb/afb-req-itf.h>
  37
  38 #include "afb-common.h"
  39
  40 #include "session.h"
  41 #include "afb-ws.h"
  42 #include "afb-msg-json.h"
  43 #include "afb-apis.h"
  44 #include "afb-api-so.h"
  45 #include "afb-context.h"
  46 #include "afb-evt.h"
  47 #include "afb-subcall.h"
  48 #include "verbose.h"
  49
  50 struct api_ws_memo;
  51 struct api_ws_event;
  52 struct api_ws_client;
  53
  54
  55
  56 /*
  57  */
  58 struct api_ws
  59 {
  60         char *path;             /* path of the object for the API */
  61         char *api;              /* api name of the interface */
  62         int fd;                 /* file descriptor */
  63         union {
  64                 struct {
  65                         uint32_t id;
  66                         struct afb_ws *ws;
  67                         struct api_ws_event *events;
  68                         struct api_ws_memo *memos;
  69                 } client;
  70                 struct {
  71                         sd_event_source *listensrc;
  72                         struct afb_evt_listener *listener; /* listener for broadcasted events */
  73                 } server;
  74         };
  75 };
  76
  77 #define RETOK   1
  78 #define RETERR  2
  79 #define RETRAW  3
  80
  81 /******************* websocket interface for client part **********************************/
  82
  83 static void api_ws_client_on_binary(void *closure, char *data, size_t size);
  84
  85 static const struct afb_ws_itf api_ws_client_ws_itf =
  86 {
  87         .on_close = NULL,
  88         .on_text = NULL,
  89         .on_binary = api_ws_client_on_binary,
  90         .on_error = NULL,
  91         .on_hangup = NULL
  92 };
  93
  94 /******************* event structures for server part **********************************/
  95
  96 static void api_ws_server_event_add(void *closure, const char *event, int eventid);
  97 static void api_ws_server_event_remove(void *closure, const char *event, int eventid);
  98 static void api_ws_server_event_push(void *closure, const char *event, int eventid, struct json_object *object);
  99 static void api_ws_server_event_broadcast(void *closure, const char *event, int eventid, struct json_object *object);
 100
 101 /* the interface for events pushing */
 102 static const struct afb_evt_itf api_ws_server_evt_itf = {
 103         .broadcast = api_ws_server_event_broadcast,
 104         .push = api_ws_server_event_push,
 105         .add = api_ws_server_event_add,
 106         .remove = api_ws_server_event_remove
 107 };
 108
 109 /******************* client description part for server *****************************/
 110
 111 struct api_ws_client
 112 {
 113         /* the server ws-api */
 114         const char *api;
 115
 116         /* count of references */
 117         int refcount;
 118
 119         /* listener for events */
 120         struct afb_evt_listener *listener;
 121
 122         /* file descriptor */
 123         int fd;
 124
 125         /* websocket */
 126         struct afb_ws *ws;
 127 };
 128
 129 /******************* websocket interface for client part **********************************/
 130
 131 static void api_ws_server_on_binary(void *closure, char *data, size_t size);
 132 static void api_ws_server_on_hangup(void *closure);
 133
 134 static const struct afb_ws_itf api_ws_server_ws_itf =
 135 {
 136         .on_close = NULL,
 137         .on_text = NULL,
 138         .on_binary = api_ws_server_on_binary,
 139         .on_error = NULL,
 140         .on_hangup = api_ws_server_on_hangup
 141 };
 142
 143 /******************* ws request part for server *****************/
 144
 145 /*
 146  * structure for a ws request
 147  */
 148 struct api_ws_server_req {
 149         struct afb_context context;     /* the context, should be THE FIRST */
 150         struct api_ws_client *client;   /* the client of the request */
 151         char *rcvdata;                  /* the received data to free */
 152         struct json_object *json;       /* the readen request as object */
 153         const char *request;            /* the readen request as string */
 154         size_t lenreq;                  /* the length of the request */
 155         int refcount;                   /* reference count of the request */
 156         uint32_t msgid;                 /* the incoming request msgid */
 157 };
 158
 159 static struct json_object *api_ws_server_req_json(struct api_ws_server_req *wreq);
 160 static struct afb_arg api_ws_server_req_get(struct api_ws_server_req *wreq, const char *name);
 161 static void api_ws_server_req_success(struct api_ws_server_req *wreq, struct json_object *obj, const char *info);
 162 static void api_ws_server_req_fail(struct api_ws_server_req *wreq, const char *status, const char *info);
 163 static const char *api_ws_server_req_raw(struct api_ws_server_req *wreq, size_t *size);
 164 static void api_ws_server_req_send(struct api_ws_server_req *wreq, const char *buffer, size_t size);
 165 static void api_ws_server_req_addref(struct api_ws_server_req *wreq);
 166 static void api_ws_server_req_unref(struct api_ws_server_req *wreq);
 167 static int api_ws_server_req_subscribe(struct api_ws_server_req *wreq, struct afb_event event);
 168 static int api_ws_server_req_unsubscribe(struct api_ws_server_req *wreq, struct afb_event event);
 169 static void api_ws_server_req_subcall(struct api_ws_server_req *wreq, const char *api, const char *verb, struct json_object *args, void (*callback)(void*, int, struct json_object*), void *closure);
 170
 171 const struct afb_req_itf afb_api_ws_req_itf = {
 172         .json = (void*)api_ws_server_req_json,
 173         .get = (void*)api_ws_server_req_get,
 174         .success = (void*)api_ws_server_req_success,
 175         .fail = (void*)api_ws_server_req_fail,
 176         .raw = (void*)api_ws_server_req_raw,
 177         .send = (void*)api_ws_server_req_send,
 178         .context_get = (void*)afb_context_get,
 179         .context_set = (void*)afb_context_set,
 180         .addref = (void*)api_ws_server_req_addref,
 181         .unref = (void*)api_ws_server_req_unref,
 182         .session_close = (void*)afb_context_close,
 183         .session_set_LOA = (void*)afb_context_change_loa,
 184         .subscribe = (void*)api_ws_server_req_subscribe,
 185         .unsubscribe = (void*)api_ws_server_req_unsubscribe,
 186         .subcall = (void*)api_ws_server_req_subcall
 187 };
 188
 189 /******************* common part **********************************/
 190
 191 /*
 192  * create a structure api_ws not connected to the 'path'.
 193  */
 194 static struct api_ws *api_ws_make(const char *path)
 195 {
 196         struct api_ws *api;
 197         size_t length;
 198
 199         /* allocates the structure */
 200         length = strlen(path);
 201         api = calloc(1, sizeof *api + 1 + length);
 202         if (api == NULL) {
 203                 errno = ENOMEM;
 204                 goto error;
 205         }
 206
 207         /* path is copied after the struct */
 208         api->path = (void*)(api+1);
 209         memcpy(api->path, path, length + 1);
 210
 211         /* api name is at the end of the path */
 212         api->api = strrchr(api->path, '/');
 213         if (api->api == NULL || !afb_apis_is_valid_api_name(++api->api)) {
 214                 errno = EINVAL;
 215                 goto error2;
 216         }
 217
 218         api->fd = -1;
 219         return api;
 220
 221 error2:
 222         free(api);
 223 error:
 224         return NULL;
 225 }
 226
 227 static int api_ws_socket_unix(const char *path, int server)
 228 {
 229         int fd, rc;
 230         struct sockaddr_un addr;
 231         size_t length;
 232
 233         length = strlen(path);
 234         if (length >= 108) {
 235                 errno = ENAMETOOLONG;
 236                 return -1;
 237         }
 238
 239         if (server)
 240                 unlink(path);
 241
 242         fd = socket(AF_UNIX, SOCK_STREAM, 0);
 243         if (fd < 0)
 244                 return fd;
 245
 246         memset(&addr, 0, sizeof addr);
 247         addr.sun_family = AF_UNIX;
 248         strcpy(addr.sun_path, path);
 249         if (server) {
 250                 rc = bind(fd, (struct sockaddr *) &addr, (socklen_t)(sizeof addr));
 251         } else {
 252                 rc = connect(fd, (struct sockaddr *) &addr, (socklen_t)(sizeof addr));
 253         }
 254         if (rc < 0) {
 255                 close(fd);
 256                 return rc;
 257         }
 258         return fd;
 259 }
 260
 261 static int api_ws_socket_inet(const char *path, int server)
 262 {
 263         int rc, fd;
 264         const char *service, *host, *api;
 265         struct addrinfo hint, *rai, *iai;
 266
 267         /* scan the uri */
 268         api = strrchr(path, '/');
 269         service = strrchr(path, ':');
 270         if (api == NULL || service == NULL || api < service) {
 271                 errno = EINVAL;
 272                 return -1;
 273         }
 274         host = strndupa(path, service++ - path);
 275         service = strndupa(service, api - service);
 276
 277         /* get addr */
 278         memset(&hint, 0, sizeof hint);
 279         hint.ai_family = AF_INET;
 280         hint.ai_socktype = SOCK_STREAM;
 281         rc = getaddrinfo(host, service, &hint, &rai);
 282         if (rc != 0) {
 283                 errno = EINVAL;
 284                 return NULL;
 285         }
 286
 287         /* get the socket */
 288         iai = rai;
 289         while (iai != NULL) {
 290                 fd = socket(iai->ai_family, iai->ai_socktype, iai->ai_protocol);
 291                 if (fd >= 0) {
 292                         if (server) {
 293                                 rc = bind(fd, iai->ai_addr, iai->ai_addrlen);
 294                         } else {
 295                                 rc = connect(fd, iai->ai_addr, iai->ai_addrlen);
 296                         }
 297                         if (rc == 0) {
 298                                 freeaddrinfo(rai);
 299                                 return fd;
 300                         }
 301                         close(fd);
 302                 }
 303                 iai = iai->ai_next;
 304         }
 305         freeaddrinfo(rai);
 306         return -1;
 307
 308 }
 309
 310 static int api_ws_socket(const char *path, int server)
 311 {
 312         int fd, rc;
 313
 314         /* check for unix socket */
 315         if (0 == strncmp(path, "unix:", 5))
 316                 fd = api_ws_socket_unix(path + 5, server);
 317         else
 318                 fd = api_ws_socket_inet(path, server);
 319
 320         if (fd >= 0) {
 321                 fcntl(fd, F_SETFD, FD_CLOEXEC);
 322                 fcntl(fd, F_SETFL, O_NONBLOCK);
 323                 if (server) {
 324                         rc = 1;
 325                         setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &rc, sizeof rc);
 326                         rc = listen(fd, 5);
 327                 }
 328         }
 329         return fd;
 330 }
 331
 332 /******************* serialisation part **********************************/
 333
 334 struct readbuf
 335 {
 336         char *head, *end;
 337 };
 338
 339 #define WRITEBUF_COUNT_MAX  32
 340 struct writebuf
 341 {
 342         struct iovec iovec[WRITEBUF_COUNT_MAX];
 343         uint32_t uints[WRITEBUF_COUNT_MAX];
 344         int count;
 345 };
 346
 347 static char *api_ws_read_get(struct readbuf *rb, uint32_t length)
 348 {
 349         char *before = rb->head;
 350         char *after = before + length;
 351         if (after > rb->end)
 352                 return 0;
 353         rb->head = after;
 354         return before;
 355 }
 356
 357 static int api_ws_read_uint32(struct readbuf *rb, uint32_t *value)
 358 {
 359         char *after = rb->head + sizeof *value;
 360         if (after > rb->end)
 361                 return 0;
 362         memcpy(value, rb->head, sizeof *value);
 363         rb->head = after;
 364         *value = le32toh(*value);
 365         return 1;
 366 }
 367
 368 static int api_ws_read_string(struct readbuf *rb, const char **value, size_t *length)
 369 {
 370         uint32_t len;
 371         if (!api_ws_read_uint32(rb, &len) || !len)
 372                 return 0;
 373         if (length)
 374                 *length = (size_t)(len - 1);
 375         return (*value = api_ws_read_get(rb, len)) != NULL &&  rb->head[-1] == 0;
 376 }
 377
 378 static int api_ws_read_object(struct readbuf *rb, struct json_object **object)
 379 {
 380         size_t length;
 381         const char *string;
 382         return api_ws_read_string(rb, &string, &length) && ((*object = json_tokener_parse(string)) != NULL) == (strcmp(string, "null") != 0);
 383 }
 384
 385 static int api_ws_write_put(struct writebuf *wb, const void *value, size_t length)
 386 {
 387         int i = wb->count;
 388         if (i == WRITEBUF_COUNT_MAX)
 389                 return 0;
 390         wb->iovec[i].iov_base = (void*)value;
 391         wb->iovec[i].iov_len = length;
 392         wb->count = i + 1;
 393         return 1;
 394 }
 395
 396 static int api_ws_write_char(struct writebuf *wb, char value)
 397 {
 398         int i = wb->count;
 399         if (i == WRITEBUF_COUNT_MAX)
 400                 return 0;
 401         *(char*)&wb->uints[i] = value;
 402         wb->iovec[i].iov_base = &wb->uints[i];
 403         wb->iovec[i].iov_len = 1;
 404         wb->count = i + 1;
 405         return 1;
 406 }
 407
 408 static int api_ws_write_uint32(struct writebuf *wb, uint32_t value)
 409 {
 410         int i = wb->count;
 411         if (i == WRITEBUF_COUNT_MAX)
 412                 return 0;
 413         wb->uints[i] = htole32(value);
 414         wb->iovec[i].iov_base = &wb->uints[i];
 415         wb->iovec[i].iov_len = sizeof wb->uints[i];
 416         wb->count = i + 1;
 417         return 1;
 418 }
 419
 420 static int api_ws_write_string_nz(struct writebuf *wb, const char *value, size_t length)
 421 {
 422         uint32_t len = (uint32_t)length;
 423         return (size_t)len == length && ++len && api_ws_write_uint32(wb, len) && api_ws_write_put(wb, value, length) && api_ws_write_char(wb, '\0');
 424 }
 425
 426 static int api_ws_write_string_length(struct writebuf *wb, const char *value, size_t length)
 427 {
 428         uint32_t len = (uint32_t)++length;
 429         return (size_t)len == length && len && api_ws_write_uint32(wb, len) && api_ws_write_put(wb, value, length);
 430 }
 431
 432 static int api_ws_write_string(struct writebuf *wb, const char *value)
 433 {
 434         return api_ws_write_string_length(wb, value, strlen(value));
 435 }
 436
 437 static int api_ws_write_object(struct writebuf *wb, struct json_object *object)
 438 {
 439         const char *string = json_object_to_json_string_ext(object, JSON_C_TO_STRING_PLAIN);
 440         return string != NULL && api_ws_write_string(wb, string);
 441 }
 442
 443
 444
 445
 446 /******************* client part **********************************/
 447
 448 /*
 449  * structure for recording query data
 450  */
 451 struct api_ws_memo {
 452         struct api_ws_memo *next;               /* the next memo */
 453         struct api_ws *api;             /* the ws api */
 454         struct afb_req req;             /* the request handle */
 455         struct afb_context *context;    /* the context of the query */
 456         uint32_t msgid;                 /* the message identifier */
 457 };
 458
 459 struct api_ws_event
 460 {
 461         struct api_ws_event *next;
 462         struct afb_event event;
 463         int eventid;
 464         int refcount;
 465 };
 466
 467 /* search a memorized request */
 468 static struct api_ws_memo *api_ws_client_memo_search(struct api_ws *api, uint32_t msgid)
 469 {
 470         struct api_ws_memo *memo;
 471
 472         memo = api->client.memos;
 473         while (memo != NULL && memo->msgid != msgid)
 474                 memo = memo->next;
 475
 476         return memo;
 477 }
 478
 479 /* search the event */
 480 static struct api_ws_event *api_ws_client_event_search(struct api_ws *api, uint32_t eventid, const char *name)
 481 {
 482         struct api_ws_event *ev;
 483
 484         ev = api->client.events;
 485         while (ev != NULL && (ev->eventid != eventid || 0 != strcmp(afb_evt_event_name(ev->event), name)))
 486                 ev = ev->next;
 487
 488         return ev;
 489 }
 490
 491
 492 /* allocates and init the memorizing data */
 493 static struct api_ws_memo *api_ws_client_memo_make(struct api_ws *api, struct afb_req req, struct afb_context *context)
 494 {
 495         struct api_ws_memo *memo;
 496
 497         memo = malloc(sizeof *memo);
 498         if (memo != NULL) {
 499                 afb_req_addref(req);
 500                 memo->req = req;
 501                 memo->context = context;
 502                 do { memo->msgid = ++api->client.id; } while(api_ws_client_memo_search(api, memo->msgid) != NULL);
 503                 memo->api = api;
 504                 memo->next = api->client.memos;
 505                 api->client.memos = memo;
 506         }
 507         return memo;
 508 }
 509
 510 /* free and release the memorizing data */
 511 static void api_ws_client_memo_destroy(struct api_ws_memo *memo)
 512 {
 513         struct api_ws_memo **prv;
 514
 515         prv = &memo->api->client.memos;
 516         while (*prv != NULL) {
 517                 if (*prv == memo) {
 518                         *prv = memo->next;
 519                         break;
 520                 }
 521                 prv = &(*prv)->next;
 522         }
 523
 524         afb_req_unref(memo->req);
 525         free(memo);
 526 }
 527
 528 /* get event data from the message */
 529 static int api_ws_client_msg_event_read(struct readbuf *rb, uint32_t *eventid, const char **name)
 530 {
 531         return api_ws_read_uint32(rb, eventid) && api_ws_read_string(rb, name, NULL);
 532 }
 533
 534 /* get event from the message */
 535 static int api_ws_client_msg_event_get(struct api_ws *api, struct readbuf *rb, struct api_ws_event **ev)
 536 {
 537         const char *name;
 538         uint32_t eventid;
 539
 540         /* get event data from the message */
 541         if (!api_ws_client_msg_event_read(rb, &eventid, &name)) {
 542                 ERROR("Invalid message");
 543                 return 0;
 544         }
 545
 546         /* check conflicts */
 547         *ev = api_ws_client_event_search(api, eventid, name);
 548         if (*ev == NULL) {
 549                 ERROR("event %s not found", name);
 550                 return 0;
 551         }
 552
 553         return 1;
 554 }
 555
 556 /* get event from the message */
 557 static int api_ws_client_msg_memo_get(struct api_ws *api, struct readbuf *rb, struct api_ws_memo **memo)
 558 {
 559         uint32_t msgid;
 560
 561         /* get event data from the message */
 562         if (!api_ws_read_uint32(rb, &msgid)) {
 563                 ERROR("Invalid message");
 564                 return 0;
 565         }
 566
 567         /* get the memo */
 568         *memo = api_ws_client_memo_search(api, msgid);
 569         if (*memo == NULL) {
 570                 ERROR("message not found");
 571                 return 0;
 572         }
 573
 574         return 1;
 575 }
 576
 577 /* read a subscrition message */
 578 static int api_ws_client_msg_subscription_get(struct api_ws *api, struct readbuf *rb, struct api_ws_event **ev, struct api_ws_memo **memo)
 579 {
 580         return api_ws_client_msg_memo_get(api, rb, memo) && api_ws_client_msg_event_get(api, rb, ev);
 581 }
 582
 583 /* adds an event */
 584 static void api_ws_client_event_create(struct api_ws *api, struct readbuf *rb)
 585 {
 586         size_t offset;
 587         const char *name;
 588         uint32_t eventid;
 589         struct api_ws_event *ev;
 590
 591         /* get event data from the message */
 592         offset = api_ws_client_msg_event_read(rb, &eventid, &name);
 593         if (offset == 0) {
 594                 ERROR("Invalid message");
 595                 return;
 596         }
 597
 598         /* check conflicts */
 599         ev = api_ws_client_event_search(api, eventid, name);
 600         if (ev != NULL) {
 601                 ev->refcount++;
 602                 return;
 603         }
 604
 605         /* no conflict, try to add it */
 606         ev = malloc(sizeof *ev);
 607         if (ev != NULL) {
 608                 ev->event = afb_evt_create_event(name);
 609                 if (ev->event.closure == NULL)
 610                         free(ev);
 611                 else {
 612                         ev->refcount = 1;
 613                         ev->eventid = eventid;
 614                         ev->next = api->client.events;
 615                         api->client.events = ev;
 616                         return;
 617                 }
 618         }
 619         ERROR("can't create event %s, out of memory", name);
 620 }
 621
 622 /* removes an event */
 623 static void api_ws_client_event_drop(struct api_ws *api, struct readbuf *rb)
 624 {
 625         struct api_ws_event *ev, **prv;
 626
 627         /* retrieves the event */
 628         if (!api_ws_client_msg_event_get(api, rb, &ev))
 629                 return;
 630
 631         /* decrease the reference count */
 632         if (--ev->refcount)
 633                 return;
 634
 635         /* unlinks the event */
 636         prv = &api->client.events;
 637         while (*prv != ev)
 638                 prv = &(*prv)->next;
 639         *prv = ev->next;
 640
 641         /* destroys the event */
 642         afb_event_drop(ev->event);
 643         free(ev);
 644 }
 645
 646 /* subscribes an event */
 647 static void api_ws_client_event_subscribe(struct api_ws *api, struct readbuf *rb)
 648 {
 649         struct api_ws_event *ev;
 650         struct api_ws_memo *memo;
 651
 652         if (api_ws_client_msg_subscription_get(api, rb, &ev, &memo)) {
 653                 /* subscribe the request from the event */
 654                 if (afb_req_subscribe(memo->req, ev->event) < 0)
 655                         ERROR("can't subscribe: %m");
 656         }
 657 }
 658
 659 /* unsubscribes an event */
 660 static void api_ws_client_event_unsubscribe(struct api_ws *api, struct readbuf *rb)
 661 {
 662         struct api_ws_event *ev;
 663         struct api_ws_memo *memo;
 664
 665         if (api_ws_client_msg_subscription_get(api, rb, &ev, &memo)) {
 666                 /* unsubscribe the request from the event */
 667                 if (afb_req_unsubscribe(memo->req, ev->event) < 0)
 668                         ERROR("can't unsubscribe: %m");
 669         }
 670 }
 671
 672 /* receives broadcasted events */
 673 static void api_ws_client_event_broadcast(struct api_ws *api, struct readbuf *rb)
 674 {
 675         struct json_object *object;
 676         const char *event;
 677
 678         if (api_ws_read_string(rb, &event, NULL) && api_ws_read_object(rb, &object))
 679                 afb_evt_broadcast(event, object);
 680         else
 681                 ERROR("unreadable broadcasted event");
 682 }
 683
 684 /* pushs an event */
 685 static void api_ws_client_event_push(struct api_ws *api, struct readbuf *rb)
 686 {
 687         struct api_ws_event *ev;
 688         struct json_object *object;
 689
 690         if (api_ws_client_msg_event_get(api, rb, &ev) && api_ws_read_object(rb, &object))
 691                 afb_event_push(ev->event, object);
 692         else
 693                 ERROR("unreadable push event");
 694 }
 695
 696 static void api_ws_client_reply_success(struct api_ws *api, struct readbuf *rb)
 697 {
 698         struct api_ws_memo *memo;
 699         struct json_object *object;
 700         const char *info;
 701         uint32_t flags;
 702
 703         /* retrieve the message data */
 704         if (!api_ws_client_msg_memo_get(api, rb, &memo))
 705                 return;
 706
 707         if (api_ws_read_uint32(rb, &flags)
 708          && api_ws_read_string(rb, &info, NULL)
 709          && api_ws_read_object(rb, &object)) {
 710                 memo->context->flags = (unsigned)flags;
 711                 afb_req_success(memo->req, object, *info ? info : NULL);
 712         } else {
 713                 /* failing to have the answer */
 714                 afb_req_fail(memo->req, "error", "ws error");
 715         }
 716         api_ws_client_memo_destroy(memo);
 717 }
 718
 719 static void api_ws_client_reply_fail(struct api_ws *api, struct readbuf *rb)
 720 {
 721         struct api_ws_memo *memo;
 722         const char *info, *status;
 723         uint32_t flags;
 724
 725         /* retrieve the message data */
 726         if (!api_ws_client_msg_memo_get(api, rb, &memo))
 727                 return;
 728
 729         if (api_ws_read_uint32(rb, &flags)
 730          && api_ws_read_string(rb, &status, NULL)
 731          && api_ws_read_string(rb, &info, NULL)) {
 732                 memo->context->flags = (unsigned)flags;
 733                 afb_req_fail(memo->req, status, *info ? info : NULL);
 734         } else {
 735                 /* failing to have the answer */
 736                 afb_req_fail(memo->req, "error", "ws error");
 737         }
 738         api_ws_client_memo_destroy(memo);
 739 }
 740
 741 static void api_ws_client_reply_send(struct api_ws *api, struct readbuf *rb)
 742 {
 743         struct api_ws_memo *memo;
 744         const char *data;
 745         size_t length;
 746         uint32_t flags;
 747
 748         /* retrieve the message data */
 749         if (!api_ws_client_msg_memo_get(api, rb, &memo))
 750                 return;
 751
 752         if (api_ws_read_uint32(rb, &flags)
 753          && api_ws_read_string(rb, &data, &length)) {
 754                 memo->context->flags = (unsigned)flags;
 755                 afb_req_send(memo->req, data, length);
 756         } else {
 757                 /* failing to have the answer */
 758                 afb_req_fail(memo->req, "error", "ws error");
 759         }
 760         api_ws_client_memo_destroy(memo);
 761 }
 762
 763 /* callback when receiving binary data */
 764 static void api_ws_client_on_binary(void *closure, char *data, size_t size)
 765 {
 766         if (size > 0) {
 767                 struct readbuf rb = { .head = data, .end = data + size };
 768                 switch (*rb.head++) {
 769                 case 'T': /* success */
 770                         api_ws_client_reply_success(closure, &rb);
 771                         break;
 772                 case 'F': /* fail */
 773                         api_ws_client_reply_fail(closure, &rb);
 774                         break;
 775                 case 'X': /* send */
 776                         api_ws_client_reply_send(closure, &rb);
 777                         break;
 778                 case '*': /* broadcast */
 779                         api_ws_client_event_broadcast(closure, &rb);
 780                         break;
 781                 case '+': /* creates the event */
 782                         api_ws_client_event_create(closure, &rb);
 783                         break;
 784                 case '-': /* drops the event */
 785                         api_ws_client_event_drop(closure, &rb);
 786                         break;
 787                 case '!': /* pushs the event */
 788                         api_ws_client_event_push(closure, &rb);
 789                         break;
 790                 case 'S': /* subscribe event for a request */
 791                         api_ws_client_event_subscribe(closure, &rb);
 792                         break;
 793                 case 'U': /* unsubscribe event for a request */
 794                         api_ws_client_event_unsubscribe(closure, &rb);
 795                         break;
 796                 default: /* unexpected message */
 797                         break;
 798                 }
 799         }
 800 }
 801
 802 /* on call, propagate it to the ws service */
 803 static void api_ws_client_call(struct api_ws *api, struct afb_req req, struct afb_context *context, const char *verb, size_t lenverb)
 804 {
 805         int rc;
 806         struct api_ws_memo *memo;
 807         struct writebuf wb = { .count = 0 };
 808         const char *raw;
 809         size_t szraw;
 810
 811         /* create the recording data */
 812         memo = api_ws_client_memo_make(api, req, context);
 813         if (memo == NULL) {
 814                 afb_req_fail(req, "error", "out of memory");
 815                 return;
 816         }
 817
 818         /* creates the call message */
 819         raw = afb_req_raw(req, &szraw);
 820         if (raw == NULL)
 821                 goto internal_error;
 822         if (!api_ws_write_uint32(&wb, memo->msgid)
 823          || !api_ws_write_uint32(&wb, (uint32_t)context->flags)
 824          || !api_ws_write_string_nz(&wb, verb, lenverb)
 825          || !api_ws_write_string(&wb, ctxClientGetUuid(context->session))
 826          || !api_ws_write_string_length(&wb, raw, szraw))
 827                 goto overflow;
 828
 829         /* send */
 830         rc = afb_ws_binary_v(api->client.ws, wb.iovec, wb.count);
 831         if (rc < 0)
 832                 goto ws_send_error;
 833         return;
 834
 835 ws_send_error:
 836         afb_req_fail(req, "error", "websocket sending error");
 837         goto clean_memo;
 838
 839 internal_error:
 840         afb_req_fail(req, "error", "internal: raw is NULL!");
 841         goto clean_memo;
 842
 843 overflow:
 844         afb_req_fail(req, "error", "overflow: size doesn't match 32 bits!");
 845
 846 clean_memo:
 847         api_ws_client_memo_destroy(memo);
 848 }
 849
 850 static int api_ws_service_start(struct api_ws *api, int share_session, int onneed)
 851 {
 852         /* not an error when onneed */
 853         if (onneed != 0)
 854                 return 0;
 855
 856         /* already started: it is an error */
 857         ERROR("The WS binding %s is not a startable service", api->path);
 858         return -1;
 859 }
 860
 861 /*  */
 862 static void api_ws_client_disconnect(struct api_ws *api)
 863 {
 864         if (api->fd >= 0) {
 865                 afb_ws_destroy(api->client.ws);
 866                 api->client.ws = NULL;
 867                 close(api->fd);
 868                 api->fd = -1;
 869         }
 870 }
 871
 872 /*  */
 873 static int api_ws_client_connect(struct api_ws *api)
 874 {
 875         struct afb_ws *ws;
 876         int fd;
 877
 878         fd = api_ws_socket(api->path, 0);
 879         if (fd >= 0) {
 880                 ws = afb_ws_create(afb_common_get_event_loop(), fd, &api_ws_client_ws_itf, api);
 881                 if (ws != NULL) {
 882                         api->client.ws = ws;
 883                         api->fd = fd;
 884                         return 0;
 885                 }
 886                 close(fd);
 887         }
 888         return -1;
 889 }
 890
 891 /* adds a afb-ws-service client api */
 892 int afb_api_ws_add_client(const char *path)
 893 {
 894         int rc;
 895         struct api_ws *api;
 896         struct afb_api afb_api;
 897
 898         /* create the ws client api */
 899         api = api_ws_make(path);
 900         if (api == NULL)
 901                 goto error;
 902
 903         /* connect to the service */
 904         rc = api_ws_client_connect(api);
 905         if (rc < 0) {
 906                 ERROR("can't connect to ws service %s", api->path);
 907                 goto error2;
 908         }
 909
 910         /* record it as an API */
 911         afb_api.closure = api;
 912         afb_api.call = (void*)api_ws_client_call;
 913         afb_api.service_start = (void*)api_ws_service_start;
 914         if (afb_apis_add(api->api, afb_api) < 0)
 915                 goto error3;
 916
 917         return 0;
 918
 919 error3:
 920         api_ws_client_disconnect(api);
 921 error2:
 922         free(api);
 923 error:
 924         return -1;
 925 }
 926
 927 /******************* client description part for server *****************************/
 928
 929 static void api_ws_server_client_unref(struct api_ws_client *client)
 930 {
 931         if (!--client->refcount) {
 932                 afb_evt_listener_unref(client->listener);
 933                 afb_ws_destroy(client->ws);
 934                 free(client);
 935         }
 936 }
 937
 938 /* on call, propagate it to the ws service */
 939 static void api_ws_server_called(struct api_ws_client *client, struct readbuf *rb, char *data, size_t size)
 940 {
 941         struct api_ws_server_req *wreq;
 942         struct afb_req areq;
 943         const char *uuid, *verb;
 944         uint32_t flags;
 945
 946         client->refcount++;
 947
 948         /* create the request */
 949         wreq = calloc(1 , sizeof *wreq);
 950         if (wreq == NULL)
 951                 goto out_of_memory;
 952
 953         wreq->client = client;
 954         wreq->rcvdata = data;
 955         wreq->refcount = 1;
 956
 957         /* reads the call message data */
 958         if (!api_ws_read_uint32(rb, &wreq->msgid)
 959          || !api_ws_read_uint32(rb, &flags)
 960          || !api_ws_read_string(rb, &verb, NULL)
 961          || !api_ws_read_string(rb, &uuid, NULL)
 962          || !api_ws_read_string(rb, &wreq->request, &wreq->lenreq))
 963                 goto overflow;
 964
 965         /* init the context */
 966         if (afb_context_connect(&wreq->context, uuid, NULL) < 0)
 967                 goto out_of_memory;
 968         wreq->context.flags = flags;
 969
 970         /* makes the call */
 971         areq.itf = &afb_api_ws_req_itf;
 972         areq.closure = wreq;
 973         afb_apis_call_(areq, &wreq->context, client->api, verb);
 974         api_ws_server_req_unref(wreq);
 975         return;
 976
 977 out_of_memory:
 978 overflow:
 979         free(wreq);
 980         free(data);
 981         api_ws_server_client_unref(client);
 982 }
 983
 984 /* callback when receiving binary data */
 985 static void api_ws_server_on_binary(void *closure, char *data, size_t size)
 986 {
 987         struct readbuf rb = { .head = data, .end = data + size };
 988         api_ws_server_called(closure, &rb, data, size);
 989 }
 990
 991 /* callback when receiving a hangup */
 992 static void api_ws_server_on_hangup(void *closure)
 993 {
 994         struct api_ws_client *client = closure;
 995
 996         /* close the socket */
 997         if (client->fd >= 0) {
 998                 close(client->fd);
 999                 client->fd = -1;
1000         }
1001
1002         /* release the client */
1003         api_ws_server_client_unref(client);
1004 }
1005
1006 static void api_ws_server_accept(struct api_ws *api)
1007 {
1008         struct api_ws_client *client;
1009         struct sockaddr addr;
1010         socklen_t lenaddr;
1011
1012         client = calloc(1, sizeof *client);
1013         if (client != NULL) {
1014                 client->listener = afb_evt_listener_create(&api_ws_server_evt_itf, client);
1015                 if (client->listener != NULL) {
1016                         lenaddr = (socklen_t)sizeof addr;
1017                         client->fd = accept(api->fd, &addr, &lenaddr);
1018                         if (client->fd >= 0) {
1019                                 client->ws = afb_ws_create(afb_common_get_event_loop(), client->fd, &api_ws_server_ws_itf, client);
1020                                 if (client->ws != NULL) {
1021                                         client->api = api->api;
1022                                         client->refcount = 1;
1023                                         return;
1024                                 }
1025                                 close(client->fd);
1026                         }
1027                         afb_evt_listener_unref(client->listener);
1028                 }
1029                 free(client);
1030         }
1031 }
1032
1033 /******************* server part: manage events **********************************/
1034
1035 static void api_ws_server_event_send(struct api_ws_client *client, char order, const char *event, int eventid, const char *data)
1036 {
1037         int rc;
1038         struct writebuf wb = { .count = 0 };
1039
1040         if (api_ws_write_char(&wb, order)
1041          && api_ws_write_uint32(&wb, eventid)
1042          && api_ws_write_string(&wb, event)
1043          && (data == NULL || api_ws_write_string(&wb, data))) {
1044                 rc = afb_ws_binary_v(client->ws, wb.iovec, wb.count);
1045                 if (rc >= 0)
1046                         return;
1047         }
1048         ERROR("error while sending %c for event %s", order, event);
1049 }
1050
1051 static void api_ws_server_event_add(void *closure, const char *event, int eventid)
1052 {
1053         api_ws_server_event_send(closure, '+', event, eventid, NULL);
1054 }
1055
1056 static void api_ws_server_event_remove(void *closure, const char *event, int eventid)
1057 {
1058         api_ws_server_event_send(closure, '-', event, eventid, NULL);
1059 }
1060
1061 static void api_ws_server_event_push(void *closure, const char *event, int eventid, struct json_object *object)
1062 {
1063         const char *data = json_object_to_json_string_ext(object, JSON_C_TO_STRING_PLAIN);
1064         api_ws_server_event_send(closure, '!', event, eventid, data ? : "null");
1065         json_object_put(object);
1066 }
1067
1068 static void api_ws_server_event_broadcast(void *closure, const char *event, int eventid, struct json_object *object)
1069 {
1070         int rc;
1071         struct api_ws_client *client = closure;
1072
1073         struct writebuf wb = { .count = 0 };
1074
1075         if (api_ws_write_char(&wb, '*') && api_ws_write_string(&wb, event) && api_ws_write_object(&wb, object)) {
1076                 rc = afb_ws_binary_v(client->ws, wb.iovec, wb.count);
1077                 if (rc < 0)
1078                         ERROR("error while broadcasting event %s", event);
1079         } else
1080                 ERROR("error while broadcasting event %s", event);
1081         json_object_put(object);
1082 }
1083
1084 /******************* ws request part for server *****************/
1085
1086 /* increment the reference count of the request */
1087 static void api_ws_server_req_addref(struct api_ws_server_req *wreq)
1088 {
1089         wreq->refcount++;
1090 }
1091
1092 /* decrement the reference count of the request and free/release it on falling to null */
1093 static void api_ws_server_req_unref(struct api_ws_server_req *wreq)
1094 {
1095         if (wreq == NULL || --wreq->refcount)
1096                 return;
1097
1098         afb_context_disconnect(&wreq->context);
1099         json_object_put(wreq->json);
1100         free(wreq->rcvdata);
1101         api_ws_server_client_unref(wreq->client);
1102         free(wreq);
1103 }
1104
1105 /* get the object of the request */
1106 static struct json_object *api_ws_server_req_json(struct api_ws_server_req *wreq)
1107 {
1108         if (wreq->json == NULL) {
1109                 wreq->json = json_tokener_parse(wreq->request);
1110                 if (wreq->json == NULL && strcmp(wreq->request, "null")) {
1111                         /* lazy error detection of json request. Is it to improve? */
1112                         wreq->json = json_object_new_string(wreq->request);
1113                 }
1114         }
1115         return wreq->json;
1116 }
1117
1118 /* get the argument of the request of 'name' */
1119 static struct afb_arg api_ws_server_req_get(struct api_ws_server_req *wreq, const char *name)
1120 {
1121         return afb_msg_json_get_arg(api_ws_server_req_json(wreq), name);
1122 }
1123
1124 static void api_ws_server_req_success(struct api_ws_server_req *wreq, struct json_object *obj, const char *info)
1125 {
1126         int rc;
1127         struct writebuf wb = { .count = 0 };
1128
1129         if (api_ws_write_char(&wb, 'T')
1130          && api_ws_write_uint32(&wb, wreq->msgid)
1131          && api_ws_write_uint32(&wb, (uint32_t)wreq->context.flags)
1132          && api_ws_write_string(&wb, info ? : "")
1133          && api_ws_write_object(&wb, obj)) {
1134                 rc = afb_ws_binary_v(wreq->client->ws, wb.iovec, wb.count);
1135                 if (rc >= 0)
1136                         goto success;
1137         }
1138         ERROR("error while sending success");
1139 success:
1140         json_object_put(obj);
1141 }
1142
1143 static void api_ws_server_req_fail(struct api_ws_server_req *wreq, const char *status, const char *info)
1144 {
1145         int rc;
1146         struct writebuf wb = { .count = 0 };
1147
1148         if (api_ws_write_char(&wb, 'F')
1149          && api_ws_write_uint32(&wb, wreq->msgid)
1150          && api_ws_write_uint32(&wb, (uint32_t)wreq->context.flags)
1151          && api_ws_write_string(&wb, status)
1152          && api_ws_write_string(&wb, info ? : "")) {
1153                 rc = afb_ws_binary_v(wreq->client->ws, wb.iovec, wb.count);
1154                 if (rc >= 0)
1155                         return;
1156         }
1157         ERROR("error while sending fail");
1158 }
1159
1160 static const char *api_ws_server_req_raw(struct api_ws_server_req *wreq, size_t *size)
1161 {
1162         if (size != NULL)
1163                 *size = wreq->lenreq;
1164         return wreq->request;
1165 }
1166
1167 static void api_ws_server_req_send(struct api_ws_server_req *wreq, const char *buffer, size_t size)
1168 {
1169         /* TODO: how to put sized buffer as strings? things aren't clear here!!! */
1170         int rc;
1171         struct writebuf wb = { .count = 0 };
1172
1173         if (api_ws_write_char(&wb, 'X')
1174          && api_ws_write_uint32(&wb, wreq->msgid)
1175          && api_ws_write_uint32(&wb, (uint32_t)wreq->context.flags)
1176          && api_ws_write_string_length(&wb, buffer, size)) {
1177                 rc = afb_ws_binary_v(wreq->client->ws, wb.iovec, wb.count);
1178                 if (rc >= 0)
1179                         return;
1180         }
1181         ERROR("error while sending raw");
1182 }
1183
1184 static int api_ws_server_req_subscribe(struct api_ws_server_req *wreq, struct afb_event event)
1185 {
1186         int rc, rc2;
1187         struct writebuf wb = { .count = 0 };
1188
1189         rc = afb_evt_add_watch(wreq->client->listener, event);
1190         if (rc < 0)
1191                 return rc;
1192
1193         if (api_ws_write_char(&wb, 'S')
1194          && api_ws_write_uint32(&wb, wreq->msgid)
1195          && api_ws_write_uint32(&wb, (uint32_t)afb_evt_event_id(event))
1196          && api_ws_write_string(&wb, afb_evt_event_name(event))) {
1197                 rc2 = afb_ws_binary_v(wreq->client->ws, wb.iovec, wb.count);
1198                 if (rc2 >= 0)
1199                         goto success;
1200         }
1201         ERROR("error while subscribing event");
1202 success:
1203         return rc;
1204 }
1205
1206 static int api_ws_server_req_unsubscribe(struct api_ws_server_req *wreq, struct afb_event event)
1207 {
1208         int rc, rc2;
1209         struct writebuf wb = { .count = 0 };
1210
1211         if (api_ws_write_char(&wb, 'U')
1212          && api_ws_write_uint32(&wb, wreq->msgid)
1213          && api_ws_write_uint32(&wb, (uint32_t)afb_evt_event_id(event))
1214          && api_ws_write_string(&wb, afb_evt_event_name(event))) {
1215                 rc2 = afb_ws_binary_v(wreq->client->ws, wb.iovec, wb.count);
1216                 if (rc2 >= 0)
1217                         goto success;
1218         }
1219         ERROR("error while subscribing event");
1220 success:
1221         rc = afb_evt_remove_watch(wreq->client->listener, event);
1222         return rc;
1223 }
1224
1225 static void api_ws_server_req_subcall(struct api_ws_server_req *wreq, const char *api, const char *verb, struct json_object *args, void (*callback)(void*, int, struct json_object*), void *closure)
1226 {
1227         afb_subcall(&wreq->context, api, verb, args, callback, closure, (struct afb_req){ .itf = &afb_api_ws_req_itf, .closure = wreq });
1228 }
1229
1230 /******************* server part **********************************/
1231
1232 static int api_ws_server_connect(struct api_ws *api);
1233
1234 static int api_ws_server_listen_callback(sd_event_source *src, int fd, uint32_t revents, void *closure)
1235 {
1236         if ((revents & EPOLLIN) != 0)
1237                 api_ws_server_accept(closure);
1238         if ((revents & EPOLLHUP) != 0)
1239                 api_ws_server_connect(closure);
1240         return 0;
1241 }
1242
1243 static void api_ws_server_disconnect(struct api_ws *api)
1244 {
1245         if (api->server.listensrc != NULL) {
1246                 sd_event_source_unref(api->server.listensrc);
1247                 api->server.listensrc = NULL;
1248         }
1249         if (api->fd >= 0) {
1250                 close(api->fd);
1251                 api->fd = -1;
1252         }
1253 }
1254
1255 static int api_ws_server_connect(struct api_ws *api)
1256 {
1257         int rc;
1258
1259         /* ensure disconnected */
1260         api_ws_server_disconnect(api);
1261
1262         /* request the service object name */
1263         api->fd = api_ws_socket(api->path, 1);
1264         if (api->fd < 0)
1265                 ERROR("can't create socket %s", api->path);
1266         else {
1267                 /* listen for service */
1268                 rc = sd_event_add_io(afb_common_get_event_loop(), &api->server.listensrc, api->fd, EPOLLIN, api_ws_server_listen_callback, api);
1269                 if (rc >= 0)
1270                         return 0;
1271                 close(api->fd);
1272                 errno = -rc;
1273                 ERROR("can't add ws object %s", api->path);
1274         }
1275         return -1;
1276 }
1277
1278 /* create the service */
1279 int afb_api_ws_add_server(const char *path)
1280 {
1281         int rc;
1282         struct api_ws *api;
1283
1284         /* creates the ws api object */
1285         api = api_ws_make(path);
1286         if (api == NULL)
1287                 goto error;
1288
1289         /* connect for serving */
1290         rc = api_ws_server_connect(api);
1291         if (rc < 0)
1292                 goto error2;
1293
1294         return 0;
1295
1296 error2:
1297         free(api);
1298 error:
1299         return -1;
1300 }
1301
1302