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