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         free(data);
 801 }
 802
 803 /* on call, propagate it to the ws service */
 804 static void api_ws_client_call(struct api_ws *api, struct afb_req req, struct afb_context *context, const char *verb, size_t lenverb)
 805 {
 806         int rc;
 807         struct api_ws_memo *memo;
 808         struct writebuf wb = { .count = 0 };
 809         const char *raw;
 810         size_t szraw;
 811
 812         /* create the recording data */
 813         memo = api_ws_client_memo_make(api, req, context);
 814         if (memo == NULL) {
 815                 afb_req_fail(req, "error", "out of memory");
 816                 return;
 817         }
 818
 819         /* creates the call message */
 820         raw = afb_req_raw(req, &szraw);
 821         if (raw == NULL)
 822                 goto internal_error;
 823         if (!api_ws_write_uint32(&wb, memo->msgid)
 824          || !api_ws_write_uint32(&wb, (uint32_t)context->flags)
 825          || !api_ws_write_string_nz(&wb, verb, lenverb)
 826          || !api_ws_write_string(&wb, ctxClientGetUuid(context->session))
 827          || !api_ws_write_string_length(&wb, raw, szraw))
 828                 goto overflow;
 829
 830         /* send */
 831         rc = afb_ws_binary_v(api->client.ws, wb.iovec, wb.count);
 832         if (rc < 0)
 833                 goto ws_send_error;
 834         return;
 835
 836 ws_send_error:
 837         afb_req_fail(req, "error", "websocket sending error");
 838         goto clean_memo;
 839
 840 internal_error:
 841         afb_req_fail(req, "error", "internal: raw is NULL!");
 842         goto clean_memo;
 843
 844 overflow:
 845         afb_req_fail(req, "error", "overflow: size doesn't match 32 bits!");
 846
 847 clean_memo:
 848         api_ws_client_memo_destroy(memo);
 849 }
 850
 851 static int api_ws_service_start(struct api_ws *api, int share_session, int onneed)
 852 {
 853         /* not an error when onneed */
 854         if (onneed != 0)
 855                 return 0;
 856
 857         /* already started: it is an error */
 858         ERROR("The WS binding %s is not a startable service", api->path);
 859         return -1;
 860 }
 861
 862 /*  */
 863 static void api_ws_client_disconnect(struct api_ws *api)
 864 {
 865         if (api->fd >= 0) {
 866                 afb_ws_destroy(api->client.ws);
 867                 api->client.ws = NULL;
 868                 close(api->fd);
 869                 api->fd = -1;
 870         }
 871 }
 872
 873 /*  */
 874 static int api_ws_client_connect(struct api_ws *api)
 875 {
 876         struct afb_ws *ws;
 877         int fd;
 878
 879         fd = api_ws_socket(api->path, 0);
 880         if (fd >= 0) {
 881                 ws = afb_ws_create(afb_common_get_event_loop(), fd, &api_ws_client_ws_itf, api);
 882                 if (ws != NULL) {
 883                         api->client.ws = ws;
 884                         api->fd = fd;
 885                         return 0;
 886                 }
 887                 close(fd);
 888         }
 889         return -1;
 890 }
 891
 892 /* adds a afb-ws-service client api */
 893 int afb_api_ws_add_client(const char *path)
 894 {
 895         int rc;
 896         struct api_ws *api;
 897         struct afb_api afb_api;
 898
 899         /* create the ws client api */
 900         api = api_ws_make(path);
 901         if (api == NULL)
 902                 goto error;
 903
 904         /* connect to the service */
 905         rc = api_ws_client_connect(api);
 906         if (rc < 0) {
 907                 ERROR("can't connect to ws service %s", api->path);
 908                 goto error2;
 909         }
 910
 911         /* record it as an API */
 912         afb_api.closure = api;
 913         afb_api.call = (void*)api_ws_client_call;
 914         afb_api.service_start = (void*)api_ws_service_start;
 915         if (afb_apis_add(api->api, afb_api) < 0)
 916                 goto error3;
 917
 918         return 0;
 919
 920 error3:
 921         api_ws_client_disconnect(api);
 922 error2:
 923         free(api);
 924 error:
 925         return -1;
 926 }
 927
 928 /******************* client description part for server *****************************/
 929
 930 static void api_ws_server_client_unref(struct api_ws_client *client)
 931 {
 932         if (!--client->refcount) {
 933                 afb_evt_listener_unref(client->listener);
 934                 afb_ws_destroy(client->ws);
 935                 free(client);
 936         }
 937 }
 938
 939 /* on call, propagate it to the ws service */
 940 static void api_ws_server_called(struct api_ws_client *client, struct readbuf *rb, char *data, size_t size)
 941 {
 942         struct api_ws_server_req *wreq;
 943         struct afb_req areq;
 944         const char *uuid, *verb;
 945         uint32_t flags;
 946
 947         client->refcount++;
 948
 949         /* create the request */
 950         wreq = calloc(1 , sizeof *wreq);
 951         if (wreq == NULL)
 952                 goto out_of_memory;
 953
 954         wreq->client = client;
 955         wreq->rcvdata = data;
 956         wreq->refcount = 1;
 957
 958         /* reads the call message data */
 959         if (!api_ws_read_uint32(rb, &wreq->msgid)
 960          || !api_ws_read_uint32(rb, &flags)
 961          || !api_ws_read_string(rb, &verb, NULL)
 962          || !api_ws_read_string(rb, &uuid, NULL)
 963          || !api_ws_read_string(rb, &wreq->request, &wreq->lenreq))
 964                 goto overflow;
 965
 966         /* init the context */
 967         if (afb_context_connect(&wreq->context, uuid, NULL) < 0)
 968                 goto out_of_memory;
 969         wreq->context.flags = flags;
 970
 971         /* makes the call */
 972         areq.itf = &afb_api_ws_req_itf;
 973         areq.closure = wreq;
 974         afb_apis_call_(areq, &wreq->context, client->api, verb);
 975         api_ws_server_req_unref(wreq);
 976         return;
 977
 978 out_of_memory:
 979 overflow:
 980         free(wreq);
 981         free(data);
 982         api_ws_server_client_unref(client);
 983 }
 984
 985 /* callback when receiving binary data */
 986 static void api_ws_server_on_binary(void *closure, char *data, size_t size)
 987 {
 988         struct readbuf rb = { .head = data, .end = data + size };
 989         api_ws_server_called(closure, &rb, data, size);
 990 }
 991
 992 /* callback when receiving a hangup */
 993 static void api_ws_server_on_hangup(void *closure)
 994 {
 995         struct api_ws_client *client = closure;
 996
 997         /* close the socket */
 998         if (client->fd >= 0) {
 999                 close(client->fd);
1000                 client->fd = -1;
1001         }
1002
1003         /* release the client */
1004         api_ws_server_client_unref(client);
1005 }
1006
1007 static void api_ws_server_accept(struct api_ws *api)
1008 {
1009         struct api_ws_client *client;
1010         struct sockaddr addr;
1011         socklen_t lenaddr;
1012
1013         client = calloc(1, sizeof *client);
1014         if (client != NULL) {
1015                 client->listener = afb_evt_listener_create(&api_ws_server_evt_itf, client);
1016                 if (client->listener != NULL) {
1017                         lenaddr = (socklen_t)sizeof addr;
1018                         client->fd = accept(api->fd, &addr, &lenaddr);
1019                         if (client->fd >= 0) {
1020                                 client->ws = afb_ws_create(afb_common_get_event_loop(), client->fd, &api_ws_server_ws_itf, client);
1021                                 if (client->ws != NULL) {
1022                                         client->api = api->api;
1023                                         client->refcount = 1;
1024                                         return;
1025                                 }
1026                                 close(client->fd);
1027                         }
1028                         afb_evt_listener_unref(client->listener);
1029                 }
1030                 free(client);
1031         }
1032 }
1033
1034 /******************* server part: manage events **********************************/
1035
1036 static void api_ws_server_event_send(struct api_ws_client *client, char order, const char *event, int eventid, const char *data)
1037 {
1038         int rc;
1039         struct writebuf wb = { .count = 0 };
1040
1041         if (api_ws_write_char(&wb, order)
1042          && api_ws_write_uint32(&wb, eventid)
1043          && api_ws_write_string(&wb, event)
1044          && (data == NULL || api_ws_write_string(&wb, data))) {
1045                 rc = afb_ws_binary_v(client->ws, wb.iovec, wb.count);
1046                 if (rc >= 0)
1047                         return;
1048         }
1049         ERROR("error while sending %c for event %s", order, event);
1050 }
1051
1052 static void api_ws_server_event_add(void *closure, const char *event, int eventid)
1053 {
1054         api_ws_server_event_send(closure, '+', event, eventid, NULL);
1055 }
1056
1057 static void api_ws_server_event_remove(void *closure, const char *event, int eventid)
1058 {
1059         api_ws_server_event_send(closure, '-', event, eventid, NULL);
1060 }
1061
1062 static void api_ws_server_event_push(void *closure, const char *event, int eventid, struct json_object *object)
1063 {
1064         const char *data = json_object_to_json_string_ext(object, JSON_C_TO_STRING_PLAIN);
1065         api_ws_server_event_send(closure, '!', event, eventid, data ? : "null");
1066         json_object_put(object);
1067 }
1068
1069 static void api_ws_server_event_broadcast(void *closure, const char *event, int eventid, struct json_object *object)
1070 {
1071         int rc;
1072         struct api_ws_client *client = closure;
1073
1074         struct writebuf wb = { .count = 0 };
1075
1076         if (api_ws_write_char(&wb, '*') && api_ws_write_string(&wb, event) && api_ws_write_object(&wb, object)) {
1077                 rc = afb_ws_binary_v(client->ws, wb.iovec, wb.count);
1078                 if (rc < 0)
1079                         ERROR("error while broadcasting event %s", event);
1080         } else
1081                 ERROR("error while broadcasting event %s", event);
1082         json_object_put(object);
1083 }
1084
1085 /******************* ws request part for server *****************/
1086
1087 /* increment the reference count of the request */
1088 static void api_ws_server_req_addref(struct api_ws_server_req *wreq)
1089 {
1090         wreq->refcount++;
1091 }
1092
1093 /* decrement the reference count of the request and free/release it on falling to null */
1094 static void api_ws_server_req_unref(struct api_ws_server_req *wreq)
1095 {
1096         if (wreq == NULL || --wreq->refcount)
1097                 return;
1098
1099         afb_context_disconnect(&wreq->context);
1100         json_object_put(wreq->json);
1101         free(wreq->rcvdata);
1102         api_ws_server_client_unref(wreq->client);
1103         free(wreq);
1104 }
1105
1106 /* get the object of the request */
1107 static struct json_object *api_ws_server_req_json(struct api_ws_server_req *wreq)
1108 {
1109         if (wreq->json == NULL) {
1110                 wreq->json = json_tokener_parse(wreq->request);
1111                 if (wreq->json == NULL && strcmp(wreq->request, "null")) {
1112                         /* lazy error detection of json request. Is it to improve? */
1113                         wreq->json = json_object_new_string(wreq->request);
1114                 }
1115         }
1116         return wreq->json;
1117 }
1118
1119 /* get the argument of the request of 'name' */
1120 static struct afb_arg api_ws_server_req_get(struct api_ws_server_req *wreq, const char *name)
1121 {
1122         return afb_msg_json_get_arg(api_ws_server_req_json(wreq), name);
1123 }
1124
1125 static void api_ws_server_req_success(struct api_ws_server_req *wreq, struct json_object *obj, const char *info)
1126 {
1127         int rc;
1128         struct writebuf wb = { .count = 0 };
1129
1130         if (api_ws_write_char(&wb, 'T')
1131          && api_ws_write_uint32(&wb, wreq->msgid)
1132          && api_ws_write_uint32(&wb, (uint32_t)wreq->context.flags)
1133          && api_ws_write_string(&wb, info ? : "")
1134          && api_ws_write_object(&wb, obj)) {
1135                 rc = afb_ws_binary_v(wreq->client->ws, wb.iovec, wb.count);
1136                 if (rc >= 0)
1137                         goto success;
1138         }
1139         ERROR("error while sending success");
1140 success:
1141         json_object_put(obj);
1142 }
1143
1144 static void api_ws_server_req_fail(struct api_ws_server_req *wreq, const char *status, const char *info)
1145 {
1146         int rc;
1147         struct writebuf wb = { .count = 0 };
1148
1149         if (api_ws_write_char(&wb, 'F')
1150          && api_ws_write_uint32(&wb, wreq->msgid)
1151          && api_ws_write_uint32(&wb, (uint32_t)wreq->context.flags)
1152          && api_ws_write_string(&wb, status)
1153          && api_ws_write_string(&wb, info ? : "")) {
1154                 rc = afb_ws_binary_v(wreq->client->ws, wb.iovec, wb.count);
1155                 if (rc >= 0)
1156                         return;
1157         }
1158         ERROR("error while sending fail");
1159 }
1160
1161 static const char *api_ws_server_req_raw(struct api_ws_server_req *wreq, size_t *size)
1162 {
1163         if (size != NULL)
1164                 *size = wreq->lenreq;
1165         return wreq->request;
1166 }
1167
1168 static void api_ws_server_req_send(struct api_ws_server_req *wreq, const char *buffer, size_t size)
1169 {
1170         /* TODO: how to put sized buffer as strings? things aren't clear here!!! */
1171         int rc;
1172         struct writebuf wb = { .count = 0 };
1173
1174         if (api_ws_write_char(&wb, 'X')
1175          && api_ws_write_uint32(&wb, wreq->msgid)
1176          && api_ws_write_uint32(&wb, (uint32_t)wreq->context.flags)
1177          && api_ws_write_string_length(&wb, buffer, size)) {
1178                 rc = afb_ws_binary_v(wreq->client->ws, wb.iovec, wb.count);
1179                 if (rc >= 0)
1180                         return;
1181         }
1182         ERROR("error while sending raw");
1183 }
1184
1185 static int api_ws_server_req_subscribe(struct api_ws_server_req *wreq, struct afb_event event)
1186 {
1187         int rc, rc2;
1188         struct writebuf wb = { .count = 0 };
1189
1190         rc = afb_evt_add_watch(wreq->client->listener, event);
1191         if (rc < 0)
1192                 return rc;
1193
1194         if (api_ws_write_char(&wb, 'S')
1195          && api_ws_write_uint32(&wb, wreq->msgid)
1196          && api_ws_write_uint32(&wb, (uint32_t)afb_evt_event_id(event))
1197          && api_ws_write_string(&wb, afb_evt_event_name(event))) {
1198                 rc2 = afb_ws_binary_v(wreq->client->ws, wb.iovec, wb.count);
1199                 if (rc2 >= 0)
1200                         goto success;
1201         }
1202         ERROR("error while subscribing event");
1203 success:
1204         return rc;
1205 }
1206
1207 static int api_ws_server_req_unsubscribe(struct api_ws_server_req *wreq, struct afb_event event)
1208 {
1209         int rc, rc2;
1210         struct writebuf wb = { .count = 0 };
1211
1212         if (api_ws_write_char(&wb, 'U')
1213          && api_ws_write_uint32(&wb, wreq->msgid)
1214          && api_ws_write_uint32(&wb, (uint32_t)afb_evt_event_id(event))
1215          && api_ws_write_string(&wb, afb_evt_event_name(event))) {
1216                 rc2 = afb_ws_binary_v(wreq->client->ws, wb.iovec, wb.count);
1217                 if (rc2 >= 0)
1218                         goto success;
1219         }
1220         ERROR("error while subscribing event");
1221 success:
1222         rc = afb_evt_remove_watch(wreq->client->listener, event);
1223         return rc;
1224 }
1225
1226 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)
1227 {
1228         afb_subcall(&wreq->context, api, verb, args, callback, closure, (struct afb_req){ .itf = &afb_api_ws_req_itf, .closure = wreq });
1229 }
1230
1231 /******************* server part **********************************/
1232
1233 static int api_ws_server_connect(struct api_ws *api);
1234
1235 static int api_ws_server_listen_callback(sd_event_source *src, int fd, uint32_t revents, void *closure)
1236 {
1237         if ((revents & EPOLLIN) != 0)
1238                 api_ws_server_accept(closure);
1239         if ((revents & EPOLLHUP) != 0)
1240                 api_ws_server_connect(closure);
1241         return 0;
1242 }
1243
1244 static void api_ws_server_disconnect(struct api_ws *api)
1245 {
1246         if (api->server.listensrc != NULL) {
1247                 sd_event_source_unref(api->server.listensrc);
1248                 api->server.listensrc = NULL;
1249         }
1250         if (api->fd >= 0) {
1251                 close(api->fd);
1252                 api->fd = -1;
1253         }
1254 }
1255
1256 static int api_ws_server_connect(struct api_ws *api)
1257 {
1258         int rc;
1259
1260         /* ensure disconnected */
1261         api_ws_server_disconnect(api);
1262
1263         /* request the service object name */
1264         api->fd = api_ws_socket(api->path, 1);
1265         if (api->fd < 0)
1266                 ERROR("can't create socket %s", api->path);
1267         else {
1268                 /* listen for service */
1269                 rc = sd_event_add_io(afb_common_get_event_loop(), &api->server.listensrc, api->fd, EPOLLIN, api_ws_server_listen_callback, api);
1270                 if (rc >= 0)
1271                         return 0;
1272                 close(api->fd);
1273                 errno = -rc;
1274                 ERROR("can't add ws object %s", api->path);
1275         }
1276         return -1;
1277 }
1278
1279 /* create the service */
1280 int afb_api_ws_add_server(const char *path)
1281 {
1282         int rc;
1283         struct api_ws *api;
1284
1285         /* creates the ws api object */
1286         api = api_ws_make(path);
1287         if (api == NULL)
1288                 goto error;
1289
1290         /* connect for serving */
1291         rc = api_ws_server_connect(api);
1292         if (rc < 0)
1293                 goto error2;
1294
1295         return 0;
1296
1297 error2:
1298         free(api);
1299 error:
1300         return -1;
1301 }
1302
1303