src/afb-ws.c

   1 /*
   2  * Copyright 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 #include <stdlib.h>
  20 #include <unistd.h>
  21 #include <stdint.h>
  22 #include <assert.h>
  23 #include <errno.h>
  24 #include <sys/uio.h>
  25 #include <string.h>
  26
  27 #include "websock.h"
  28 #include "afb-ws.h"
  29
  30 #include "utils-upoll.h"
  31
  32 /*
  33  * declaration of the websock interface for afb-ws
  34  */
  35 static ssize_t aws_writev(struct afb_ws *ws, const struct iovec *iov, int iovcnt);
  36 static ssize_t aws_readv(struct afb_ws *ws, const struct iovec *iov, int iovcnt);
  37 static void aws_on_close(struct afb_ws *ws, uint16_t code, size_t size);
  38 static void aws_on_text(struct afb_ws *ws, int last, size_t size);
  39 static void aws_on_binary(struct afb_ws *ws, int last, size_t size);
  40 static void aws_on_continue(struct afb_ws *ws, int last, size_t size);
  41 static void aws_on_readable(struct afb_ws *ws);
  42 static void aws_on_error(struct afb_ws *ws, uint16_t code, const void *data, size_t size);
  43
  44 static struct websock_itf aws_itf = {
  45         .writev = (void*)aws_writev,
  46         .readv = (void*)aws_readv,
  47
  48         .on_ping = NULL,
  49         .on_pong = NULL,
  50         .on_close = (void*)aws_on_close,
  51         .on_text = (void*)aws_on_text,
  52         .on_binary = (void*)aws_on_binary,
  53         .on_continue = (void*)aws_on_continue,
  54         .on_extension = NULL,
  55
  56         .on_error = (void*)aws_on_error
  57 };
  58
  59 /*
  60  * a common scheme of buffer handling
  61  */
  62 struct buf
  63 {
  64         char *buffer;
  65         size_t size;
  66 };
  67
  68 /*
  69  * the state
  70  */
  71 enum state
  72 {
  73         waiting,
  74         reading_text,
  75         reading_binary
  76 };
  77
  78 /*
  79  * the afb_ws structure
  80  */
  81 struct afb_ws
  82 {
  83         int fd;                 /* the socket file descriptor */
  84         enum state state;       /* current state */
  85         const struct afb_ws_itf *itf; /* the callback interface */
  86         void *closure;          /* closure when calling the callbacks */
  87         struct websock *ws;     /* the websock handler */
  88         struct upoll *up;       /* the upoll handler for the socket */
  89         struct buf buffer;      /* the last read fragment */
  90 };
  91
  92 /*
  93  * Returns the current buffer of 'ws' that is reset.
  94  */
  95 static inline struct buf aws_pick_buffer(struct afb_ws *ws)
  96 {
  97         struct buf result = ws->buffer;
  98         ws->buffer.buffer = NULL;
  99         ws->buffer.size = 0;
 100         return result;
 101 }
 102
 103 /*
 104  * Disconnect the websocket 'ws' and calls on_hangup if
 105  * 'call_on_hangup' is not null.
 106  */
 107 static void aws_disconnect(struct afb_ws *ws, int call_on_hangup)
 108 {
 109         struct websock *wsi = ws->ws;
 110         if (wsi != NULL) {
 111                 ws->ws = NULL;
 112                 upoll_close(ws->up);
 113                 ws->up = NULL;
 114                 websock_destroy(wsi);
 115                 free(aws_pick_buffer(ws).buffer);
 116                 ws->state = waiting;
 117                 if (call_on_hangup && ws->itf->on_hangup)
 118                         ws->itf->on_hangup(ws->closure);
 119         }
 120 }
 121
 122 /*
 123  * Creates the afb_ws structure for the file descritor
 124  * 'fd' and the callbacks described by the interface 'itf'
 125  * and its 'closure'.
 126  *
 127  * Returns the handle for the afb_ws created or NULL on error.
 128  */
 129 struct afb_ws *afb_ws_create(int fd, const struct afb_ws_itf *itf, void *closure)
 130 {
 131         struct afb_ws *result;
 132
 133         assert(fd >= 0);
 134
 135         /* allocation */
 136         result = malloc(sizeof * result);
 137         if (result == NULL)
 138                 goto error;
 139
 140         /* init */
 141         result->fd = fd;
 142         result->state = waiting;
 143         result->itf = itf;
 144         result->closure = closure;
 145         result->buffer.buffer = NULL;
 146         result->buffer.size = 0;
 147
 148         /* creates the websocket */
 149         result->ws = websock_create_v13(&aws_itf, result);
 150         if (result->ws == NULL)
 151                 goto error2;
 152
 153         /* creates the upoll */
 154         result->up = upoll_open(result->fd, result);
 155         if (result->up == NULL)
 156                 goto error3;
 157
 158         /* init the upoll */
 159         upoll_on_readable(result->up, (void*)aws_on_readable);
 160         upoll_on_hangup(result->up, (void*)afb_ws_hangup);
 161
 162         return result;
 163
 164 error3:
 165         websock_destroy(result->ws);
 166 error2:
 167         free(result);
 168 error:
 169         return NULL;
 170 }
 171
 172 /*
 173  * Destroys the websocket 'ws'
 174  * It first hangup (but without calling on_hangup for safety reasons)
 175  * if needed.
 176  */
 177 void afb_ws_destroy(struct afb_ws *ws)
 178 {
 179         aws_disconnect(ws, 0);
 180         free(ws);
 181 }
 182
 183 /*
 184  * Hangup the websocket 'ws'
 185  */
 186 void afb_ws_hangup(struct afb_ws *ws)
 187 {
 188         aws_disconnect(ws, 1);
 189 }
 190
 191 /*
 192  * Sends a 'close' command to the endpoint of 'ws' with the 'code' and the
 193  * 'reason' (that can be NULL and that else should not be greater than 123
 194  * characters).
 195  * Returns 0 on success or -1 in case of error.
 196  */
 197 int afb_ws_close(struct afb_ws *ws, uint16_t code, const char *reason)
 198 {
 199         if (ws->ws == NULL) {
 200                 /* disconnected */
 201                 errno = EPIPE;
 202                 return -1;
 203         }
 204         return websock_close(ws->ws, code, reason, reason == NULL ? 0 : strlen(reason));
 205 }
 206
 207 /*
 208  * Sends a 'close' command to the endpoint of 'ws' with the 'code' and the
 209  * 'reason' (that can be NULL and that else should not be greater than 123
 210  * characters).
 211  * Raise an error after 'close' command is sent.
 212  * Returns 0 on success or -1 in case of error.
 213  */
 214 int afb_ws_error(struct afb_ws *ws, uint16_t code, const char *reason)
 215 {
 216         if (ws->ws == NULL) {
 217                 /* disconnected */
 218                 errno = EPIPE;
 219                 return -1;
 220         }
 221         return websock_error(ws->ws, code, reason, reason == NULL ? 0 : strlen(reason));
 222 }
 223
 224 /*
 225  * Sends a 'text' of 'length' to the endpoint of 'ws'.
 226  * Returns 0 on success or -1 in case of error.
 227  */
 228 int afb_ws_text(struct afb_ws *ws, const char *text, size_t length)
 229 {
 230         if (ws->ws == NULL) {
 231                 /* disconnected */
 232                 errno = EPIPE;
 233                 return -1;
 234         }
 235         return websock_text(ws->ws, 1, text, length);
 236 }
 237
 238 /*
 239  * Sends a binary 'data' of 'length' to the endpoint of 'ws'.
 240  * Returns 0 on success or -1 in case of error.
 241  */
 242 int afb_ws_binary(struct afb_ws *ws, const void *data, size_t length)
 243 {
 244         if (ws->ws == NULL) {
 245                 /* disconnected */
 246                 errno = EPIPE;
 247                 return -1;
 248         }
 249         return websock_binary(ws->ws, 1, data, length);
 250 }
 251
 252 /*
 253  * callback for writing data
 254  */
 255 static ssize_t aws_writev(struct afb_ws *ws, const struct iovec *iov, int iovcnt)
 256 {
 257         ssize_t rc;
 258         do {
 259                 rc = writev(ws->fd, iov, iovcnt);
 260         } while(rc == -1 && errno == EINTR);
 261         return rc;
 262 }
 263
 264 /*
 265  * callback for reading data
 266  */
 267 static ssize_t aws_readv(struct afb_ws *ws, const struct iovec *iov, int iovcnt)
 268 {
 269         ssize_t rc;
 270         do {
 271                 rc = readv(ws->fd, iov, iovcnt);
 272         } while(rc == -1 && errno == EINTR);
 273         if (rc == 0) {
 274                 errno = EPIPE;
 275                 rc = -1;
 276         }
 277         return rc;
 278 }
 279
 280 /*
 281  * callback on incoming data
 282  */
 283 static void aws_on_readable(struct afb_ws *ws)
 284 {
 285         int rc;
 286
 287         assert(ws->ws != NULL);
 288         rc = websock_dispatch(ws->ws);
 289         if (rc < 0 && errno == EPIPE)
 290                 afb_ws_hangup(ws);
 291 }
 292
 293 /*
 294  * Reads from the websocket handled by 'ws' data of length 'size'
 295  * and append it to the current buffer of 'ws'.
 296  * Returns 0 in case of error or 1 in case of success.
 297  */
 298 static int aws_read(struct afb_ws *ws, size_t size)
 299 {
 300         ssize_t sz;
 301         char *buffer;
 302
 303         if (size != 0) {
 304                 buffer = realloc(ws->buffer.buffer, ws->buffer.size + size + 1);
 305                 if (buffer == NULL)
 306                         return 0;
 307                 ws->buffer.buffer = buffer;
 308                 sz = websock_read(ws->ws, &buffer[ws->buffer.size], size);
 309                 if ((size_t)sz != size)
 310                         return 0;
 311                 ws->buffer.size += size;
 312         }
 313         return 1;
 314 }
 315
 316 /*
 317  * Callback when 'close' command received from 'ws' with 'code' and 'size'.
 318  */
 319 static void aws_on_close(struct afb_ws *ws, uint16_t code, size_t size)
 320 {
 321         struct buf b;
 322
 323         ws->state = waiting;
 324         free(aws_pick_buffer(ws).buffer);
 325         if (ws->itf->on_close == NULL) {
 326                 websock_drop(ws->ws);
 327                 afb_ws_hangup(ws);
 328         } else if (!aws_read(ws, size))
 329                 ws->itf->on_close(ws->closure, code, NULL, 0);
 330         else {
 331                 b = aws_pick_buffer(ws);
 332                 ws->itf->on_close(ws->closure, code, b.buffer, b.size);
 333         }
 334 }
 335
 336 /*
 337  * Drops any incoming data and send an error of 'code'
 338  */
 339 static void aws_drop_error(struct afb_ws *ws, uint16_t code)
 340 {
 341         ws->state = waiting;
 342         free(aws_pick_buffer(ws).buffer);
 343         websock_drop(ws->ws);
 344         websock_error(ws->ws, code, NULL, 0);
 345 }
 346
 347 /*
 348  * Reads either text or binary data of 'size' from 'ws' eventually 'last'.
 349  */
 350 static void aws_continue(struct afb_ws *ws, int last, size_t size)
 351 {
 352         struct buf b;
 353         int istxt;
 354
 355         if (!aws_read(ws, size))
 356                 aws_drop_error(ws, WEBSOCKET_CODE_ABNORMAL);
 357         else if (last) {
 358                 istxt = ws->state == reading_text;
 359                 ws->state = waiting;
 360                 b = aws_pick_buffer(ws);
 361                 b.buffer[b.size] = 0;
 362                 (istxt ? ws->itf->on_text : ws->itf->on_binary)(ws->closure, b.buffer, b.size);
 363         }
 364 }
 365
 366 /*
 367  * Callback when 'text' message received from 'ws' with 'size' and possibly 'last'.
 368  */
 369 static void aws_on_text(struct afb_ws *ws, int last, size_t size)
 370 {
 371         if (ws->state != waiting)
 372                 aws_drop_error(ws, WEBSOCKET_CODE_PROTOCOL_ERROR);
 373         else if (ws->itf->on_text == NULL)
 374                 aws_drop_error(ws, WEBSOCKET_CODE_CANT_ACCEPT);
 375         else {
 376                 ws->state = reading_text;
 377                 aws_continue(ws, last, size);
 378         }
 379 }
 380
 381 /*
 382  * Callback when 'binary' message received from 'ws' with 'size' and possibly 'last'.
 383  */
 384 static void aws_on_binary(struct afb_ws *ws, int last, size_t size)
 385 {
 386         if (ws->state != waiting)
 387                 aws_drop_error(ws, WEBSOCKET_CODE_PROTOCOL_ERROR);
 388         else if (ws->itf->on_binary == NULL)
 389                 aws_drop_error(ws, WEBSOCKET_CODE_CANT_ACCEPT);
 390         else {
 391                 ws->state = reading_binary;
 392                 aws_continue(ws, last, size);
 393         }
 394 }
 395
 396 /*
 397  * Callback when 'close' command received from 'ws' with 'code' and 'size'.
 398  */
 399 static void aws_on_continue(struct afb_ws *ws, int last, size_t size)
 400 {
 401         if (ws->state == waiting)
 402                 aws_drop_error(ws, WEBSOCKET_CODE_PROTOCOL_ERROR);
 403         else
 404                 aws_continue(ws, last, size);
 405 }
 406
 407 /*
 408  * Callback when 'close' command is sent to 'ws' with 'code' and 'size'.
 409  */
 410 static void aws_on_error(struct afb_ws *ws, uint16_t code, const void *data, size_t size)
 411 {
 412         if (ws->itf->on_error != NULL)
 413                 ws->itf->on_error(ws->closure, code, data, size);
 414         else
 415                 afb_ws_hangup(ws);
 416 }
 417
 418
 419