libustcomm/ustcomm.c

   1 /* Copyright (C) 2009  Pierre-Marc Fournier
   2  *
   3  * This library is free software; you can redistribute it and/or
   4  * modify it under the terms of the GNU Lesser General Public
   5  * License as published by the Free Software Foundation; either
   6  * version 2.1 of the License, or (at your option) any later version.
   7  *
   8  * This library is distributed in the hope that it will be useful,
   9  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  10  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  11  * Lesser General Public License for more details.
  12  *
  13  * You should have received a copy of the GNU Lesser General Public
  14  * License along with this library; if not, write to the Free Software
  15  * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301 USA
  16  */
  17
  18 #define _GNU_SOURCE
  19 #include <sys/types.h>
  20 #include <signal.h>
  21 #include <errno.h>
  22 #include <sys/socket.h>
  23 #include <sys/un.h>
  24 #include <unistd.h>
  25 #include <poll.h>
  26 #include <sys/stat.h>
  27
  28 #include <stdio.h>
  29 #include <stdlib.h>
  30 #include <string.h>
  31 #include <execinfo.h>
  32
  33 #include "ustcomm.h"
  34 #include "usterr.h"
  35 #include "share.h"
  36 #include "multipoll.h"
  37
  38 #define UNIX_PATH_MAX 108
  39
  40 #define MSG_MAX 10000
  41
  42 /* FIXME: ustcomm blocks on message sending, which might be problematic in
  43  * some cases. Fix the poll() usage so sends are buffered until they don't
  44  * block.
  45  */
  46
  47 //static void bt(void)
  48 //{
  49 //      void *buffer[100];
  50 //      int result;
  51 //
  52 //      result = backtrace(&buffer, 100);
  53 //      backtrace_symbols_fd(buffer, result, STDERR_FILENO);
  54 //}
  55
  56 static int mkdir_p(const char *path, mode_t mode)
  57 {
  58         const char *path_p;
  59         char *tmp;
  60
  61         int retval = 0;
  62         int result;
  63
  64         tmp = malloc(strlen(path) + 1);
  65         if (tmp == NULL)
  66                 return -1;
  67
  68         /* skip first / */
  69         path_p = path+1;
  70
  71         for(;;) {
  72                 while (*path_p != '/') {
  73                         if(*path_p == 0)
  74                                 break;
  75                         ++path_p;
  76                 }
  77                 if (*path_p == '/') {
  78                         strncpy(tmp, path, path_p - path);
  79                         tmp[path_p-path] = '\0';
  80                         if (tmp[path_p - path - 1] != '/') {
  81                                 result = mkdir(tmp, mode);
  82                                 if(result == -1) {
  83                                         if (!(errno == EEXIST || errno == EACCES || errno == EROFS)) {
  84                                                 /* Then this is a real error */
  85                                                 retval = -1;
  86                                                 break;
  87                                         }
  88                                 }
  89                         }
  90                         /* pass / */
  91                         path_p++;
  92                 } else {
  93                         /* last component */
  94                         result = mkdir(path, mode);
  95                         if (result == -1)
  96                                 retval = -1;
  97                         break;
  98                 }
  99         }
 100
 101         free(tmp);
 102         return retval;
 103 }
 104
 105 char *strdup_malloc(const char *s)
 106 {
 107         char *retval;
 108
 109         if(s == NULL)
 110                 return NULL;
 111
 112         retval = (char *) malloc(strlen(s)+1);
 113
 114         strcpy(retval, s);
 115
 116         return retval;
 117 }
 118
 119 static int signal_process(pid_t pid)
 120 {
 121         return 0;
 122 }
 123
 124 void ustcomm_init_connection(struct ustcomm_connection *conn)
 125 {
 126         conn->recv_buf = NULL;
 127         conn->recv_buf_size = 0;
 128         conn->recv_buf_alloc = 0;
 129 }
 130
 131 int pid_is_online(pid_t pid) {
 132         return 1;
 133 }
 134
 135 /* Send a message
 136  *
 137  * @fd: file descriptor to send to
 138  * @msg: a null-terminated string containing the message to send
 139  *
 140  * Return value:
 141  * -1: error
 142  * 0: connection closed
 143  * 1: success
 144  */
 145
 146 static int send_message_fd(int fd, const char *msg)
 147 {
 148         int result;
 149
 150         /* Send including the final \0 */
 151         result = patient_send(fd, msg, strlen(msg)+1, MSG_NOSIGNAL);
 152         if(result == -1) {
 153                 if(errno != EPIPE)
 154                         PERROR("send");
 155                 return -1;
 156         }
 157         else if(result == 0) {
 158                 return 0;
 159         }
 160
 161         DBG("sent message \"%s\"", msg);
 162         return 1;
 163 }
 164
 165 /* Called by an app to ask the consumer daemon to connect to it. */
 166
 167 int ustcomm_request_consumer(pid_t pid, const char *channel)
 168 {
 169         char path[UNIX_PATH_MAX];
 170         int result;
 171         char *msg=NULL;
 172         int retval = 0;
 173         struct ustcomm_connection conn;
 174         char *explicit_daemon_socket_path;
 175
 176         explicit_daemon_socket_path = getenv("UST_DAEMON_SOCKET");
 177         if(explicit_daemon_socket_path) {
 178                 /* user specified explicitly a socket path */
 179                 result = snprintf(path, UNIX_PATH_MAX, "%s", explicit_daemon_socket_path);
 180         }
 181         else {
 182                 /* just use the default path */
 183                 result = snprintf(path, UNIX_PATH_MAX, "%s/ustd", SOCK_DIR);
 184         }
 185
 186         if(result >= UNIX_PATH_MAX) {
 187                 ERR("string overflow allocating socket name");
 188                 return -1;
 189         }
 190
 191         asprintf(&msg, "collect %d %s", pid, channel);
 192
 193         /* don't signal it because it's the daemon */
 194         result = ustcomm_connect_path(path, &conn, -1);
 195         if(result == -1) {
 196                 WARN("ustcomm_connect_path failed");
 197                 retval = -1;
 198                 goto del_string;
 199         }
 200
 201         result = ustcomm_send_request(&conn, msg, NULL);
 202         if(result == -1) {
 203                 WARN("ustcomm_send_request failed");
 204                 retval = -1;
 205                 goto disconnect;
 206         }
 207
 208         disconnect:
 209         ustcomm_disconnect(&conn);
 210         del_string:
 211         free(msg);
 212
 213         return retval;
 214 }
 215
 216 /* returns 1 to indicate a message was received
 217  * returns 0 to indicate no message was received (end of stream)
 218  * returns -1 to indicate an error
 219  */
 220
 221 #define RECV_INCREMENT 1000
 222 #define RECV_INITIAL_BUF_SIZE 10
 223
 224 static int recv_message_fd(int fd, char **recv_buf, int *recv_buf_size, int *recv_buf_alloc, char **msg)
 225 {
 226         int result;
 227
 228         /* 1. Check if there is a message in the buf */
 229         /* 2. If not, do:
 230            2.1 receive chunk and put it in buffer
 231            2.2 process full message if there is one
 232            -- while no message arrived
 233         */
 234
 235         for(;;) {
 236                 int i;
 237                 int nulfound = 0;
 238
 239                 /* Search for full message in buffer */
 240                 for(i=0; i<*recv_buf_size; i++) {
 241                         if((*recv_buf)[i] == '\0') {
 242                                 nulfound = 1;
 243                                 break;
 244                         }
 245                 }
 246
 247                 /* Process found message */
 248                 if(nulfound == 1) {
 249                         char *newbuf;
 250
 251                         if(i == 0) {
 252                                 /* problem */
 253                                 WARN("received empty message");
 254                         }
 255                         *msg = strndup(*recv_buf, i);
 256
 257                         /* Remove processed message from buffer */
 258                         newbuf = (char *) malloc(*recv_buf_size - (i+1));
 259                         memcpy(newbuf, *recv_buf + (i+1), *recv_buf_size - (i+1));
 260                         free(*recv_buf);
 261                         *recv_buf = newbuf;
 262                         *recv_buf_size -= (i+1);
 263                         *recv_buf_alloc -= (i+1);
 264
 265                         return 1;
 266                 }
 267
 268                 /* Receive a chunk from the fd */
 269                 if(*recv_buf_alloc - *recv_buf_size < RECV_INCREMENT) {
 270                         *recv_buf_alloc += RECV_INCREMENT - (*recv_buf_alloc - *recv_buf_size);
 271                         *recv_buf = (char *) realloc(*recv_buf, *recv_buf_alloc);
 272                 }
 273
 274                 result = recv(fd, *recv_buf+*recv_buf_size, RECV_INCREMENT, 0);
 275                 if(result == -1) {
 276                         if(errno == ECONNRESET) {
 277                                 *recv_buf_size = 0;
 278                                 return 0;
 279                         }
 280                         /* real error */
 281                         PERROR("recv");
 282                         return -1;
 283                 }
 284                 if(result == 0) {
 285                         return 0;
 286                 }
 287                 *recv_buf_size += result;
 288
 289                 /* Go back to the beginning to check if there is a full message in the buffer */
 290         }
 291
 292         DBG("received message \"%s\"", *recv_buf);
 293
 294         return 1;
 295
 296 }
 297
 298 static int recv_message_conn(struct ustcomm_connection *conn, char **msg)
 299 {
 300         return recv_message_fd(conn->fd, &conn->recv_buf, &conn->recv_buf_size, &conn->recv_buf_alloc, msg);
 301 }
 302
 303 int ustcomm_send_reply(struct ustcomm_server *server, char *msg, struct ustcomm_source *src)
 304 {
 305         int result;
 306
 307         result = send_message_fd(src->fd, msg);
 308         if(result < 0) {
 309                 ERR("error in send_message_fd");
 310                 return -1;
 311         }
 312
 313         return 0;
 314 }
 315
 316 /* Called after a fork. */
 317
 318 int ustcomm_close_all_connections(struct ustcomm_server *server)
 319 {
 320         struct ustcomm_connection *conn;
 321         struct ustcomm_connection *deletable_conn = NULL;
 322
 323         list_for_each_entry(conn, &server->connections, list) {
 324                 free(deletable_conn);
 325                 deletable_conn = conn;
 326                 ustcomm_close_app(conn);
 327                 list_del(&conn->list);
 328         }
 329
 330         return 0;
 331 }
 332
 333 /* @timeout: max blocking time in milliseconds, -1 means infinity
 334  *
 335  * returns 1 to indicate a message was received
 336  * returns 0 to indicate no message was received
 337  * returns -1 to indicate an error
 338  */
 339
 340 int ustcomm_recv_message(struct ustcomm_server *server, char **msg, struct ustcomm_source *src, int timeout)
 341 {
 342         struct pollfd *fds;
 343         struct ustcomm_connection **conn_table;
 344         struct ustcomm_connection *conn;
 345         int result;
 346         int retval;
 347
 348         for(;;) {
 349                 int idx = 0;
 350                 int n_fds = 1;
 351
 352                 list_for_each_entry(conn, &server->connections, list) {
 353                         n_fds++;
 354                 }
 355
 356                 fds = (struct pollfd *) malloc(n_fds * sizeof(struct pollfd));
 357                 if(fds == NULL) {
 358                         ERR("malloc returned NULL");
 359                         return -1;
 360                 }
 361
 362                 conn_table = (struct ustcomm_connection **) malloc(n_fds * sizeof(struct ustcomm_connection *));
 363                 if(conn_table == NULL) {
 364                         ERR("malloc returned NULL");
 365                         retval = -1;
 366                         goto free_fds_return;
 367                 }
 368
 369                 /* special idx 0 is for listening socket */
 370                 fds[idx].fd = server->listen_fd;
 371                 fds[idx].events = POLLIN;
 372                 idx++;
 373
 374                 list_for_each_entry(conn, &server->connections, list) {
 375                         fds[idx].fd = conn->fd;
 376                         fds[idx].events = POLLIN;
 377                         conn_table[idx] = conn;
 378                         idx++;
 379                 }
 380
 381                 while((result = poll(fds, n_fds, timeout)) == -1 && errno == EINTR)
 382                         /* nothing */;
 383                 if(result == -1) {
 384                         PERROR("poll");
 385                         retval = -1;
 386                         goto free_conn_table_return;
 387                 }
 388
 389                 if(result == 0) {
 390                         retval = 0;
 391                         goto free_conn_table_return;
 392                 }
 393
 394                 if(fds[0].revents) {
 395                         struct ustcomm_connection *newconn;
 396                         int newfd;
 397
 398                         result = newfd = accept(server->listen_fd, NULL, NULL);
 399                         if(result == -1) {
 400                                 PERROR("accept");
 401                                 retval = -1;
 402                                 goto free_conn_table_return;
 403                         }
 404
 405                         newconn = (struct ustcomm_connection *) malloc(sizeof(struct ustcomm_connection));
 406                         if(newconn == NULL) {
 407                                 ERR("malloc returned NULL");
 408                                 return -1;
 409                         }
 410
 411                         ustcomm_init_connection(newconn);
 412                         newconn->fd = newfd;
 413
 414                         list_add(&newconn->list, &server->connections);
 415                 }
 416
 417                 for(idx=1; idx<n_fds; idx++) {
 418                         if(fds[idx].revents) {
 419                                 retval = recv_message_conn(conn_table[idx], msg);
 420                                 if(src)
 421                                         src->fd = fds[idx].fd;
 422
 423                                 if(retval == 0) {
 424                                         /* connection finished */
 425                                         list_for_each_entry(conn, &server->connections, list) {
 426                                                 if(conn->fd == fds[idx].fd) {
 427                                                         ustcomm_close_app(conn);
 428                                                         list_del(&conn->list);
 429                                                         free(conn);
 430                                                         break;
 431                                                 }
 432                                         }
 433                                 }
 434                                 else {
 435                                         goto free_conn_table_return;
 436                                 }
 437                         }
 438                 }
 439
 440                 free(fds);
 441                 free(conn_table);
 442         }
 443
 444 free_conn_table_return:
 445         free(conn_table);
 446 free_fds_return:
 447         free(fds);
 448         return retval;
 449 }
 450
 451 int ustcomm_ustd_recv_message(struct ustcomm_ustd *ustd, char **msg, struct ustcomm_source *src, int timeout)
 452 {
 453         return ustcomm_recv_message(&ustd->server, msg, src, timeout);
 454 }
 455
 456 int ustcomm_app_recv_message(struct ustcomm_app *app, char **msg, struct ustcomm_source *src, int timeout)
 457 {
 458         return ustcomm_recv_message(&app->server, msg, src, timeout);
 459 }
 460
 461 /* This removes src from the list of active connections of app.
 462  */
 463
 464 int ustcomm_app_detach_client(struct ustcomm_app *app, struct ustcomm_source *src)
 465 {
 466         struct ustcomm_server *server = (struct ustcomm_server *)app;
 467         struct ustcomm_connection *conn;
 468
 469         list_for_each_entry(conn, &server->connections, list) {
 470                 if(conn->fd == src->fd) {
 471                         list_del(&conn->list);
 472                         goto found;
 473                 }
 474         }
 475
 476         return -1;
 477 found:
 478         return src->fd;
 479 }
 480
 481 static int init_named_socket(const char *name, char **path_out)
 482 {
 483         int result;
 484         int fd;
 485
 486         struct sockaddr_un addr;
 487
 488         result = fd = socket(PF_UNIX, SOCK_STREAM, 0);
 489         if(result == -1) {
 490                 PERROR("socket");
 491                 return -1;
 492         }
 493
 494         addr.sun_family = AF_UNIX;
 495
 496         strncpy(addr.sun_path, name, UNIX_PATH_MAX);
 497         addr.sun_path[UNIX_PATH_MAX-1] = '\0';
 498
 499         result = access(name, F_OK);
 500         if(result == 0) {
 501                 /* file exists */
 502                 result = unlink(name);
 503                 if(result == -1) {
 504                         PERROR("unlink of socket file");
 505                         goto close_sock;
 506                 }
 507                 DBG("socket already exists; overwriting");
 508         }
 509
 510         result = bind(fd, (struct sockaddr *)&addr, sizeof(addr));
 511         if(result == -1) {
 512                 PERROR("bind");
 513                 goto close_sock;
 514         }
 515
 516         result = listen(fd, 1);
 517         if(result == -1) {
 518                 PERROR("listen");
 519                 goto close_sock;
 520         }
 521
 522         if(path_out) {
 523                 *path_out = strdup(addr.sun_path);
 524         }
 525
 526         return fd;
 527
 528         close_sock:
 529         close(fd);
 530
 531         return -1;
 532 }
 533
 534 /*
 535  * Return value:
 536  *   0: Success, but no reply because recv() returned 0
 537  *   1: Success
 538  *   -1: Error
 539  *
 540  * On error, the error message is printed, except on
 541  * ECONNRESET, which is normal when the application dies.
 542  */
 543
 544 int ustcomm_send_request(struct ustcomm_connection *conn, const char *req, char **reply)
 545 {
 546         int result;
 547
 548         /* Send including the final \0 */
 549         result = send_message_fd(conn->fd, req);
 550         if(result != 1)
 551                 return result;
 552
 553         if(!reply)
 554                 return 1;
 555
 556         result = recv_message_conn(conn, reply);
 557         if(result == -1) {
 558                 return -1;
 559         }
 560         else if(result == 0) {
 561                 return 0;
 562         }
 563
 564         return 1;
 565 }
 566
 567 /* Return value:
 568  *  0: success
 569  * -1: error
 570  */
 571
 572 int ustcomm_connect_path(const char *path, struct ustcomm_connection *conn, pid_t signalpid)
 573 {
 574         int fd;
 575         int result;
 576         struct sockaddr_un addr;
 577
 578         ustcomm_init_connection(conn);
 579
 580         result = fd = socket(PF_UNIX, SOCK_STREAM, 0);
 581         if(result == -1) {
 582                 PERROR("socket");
 583                 return -1;
 584         }
 585
 586         addr.sun_family = AF_UNIX;
 587
 588         result = snprintf(addr.sun_path, UNIX_PATH_MAX, "%s", path);
 589         if(result >= UNIX_PATH_MAX) {
 590                 ERR("string overflow allocating socket name");
 591                 return -1;
 592         }
 593
 594         if(signalpid >= 0) {
 595                 result = signal_process(signalpid);
 596                 if(result == -1) {
 597                         ERR("could not signal process");
 598                         return -1;
 599                 }
 600         }
 601
 602         result = connect(fd, (struct sockaddr *)&addr, sizeof(addr));
 603         if(result == -1) {
 604                 PERROR("connect (path=%s)", path);
 605                 return -1;
 606         }
 607
 608         conn->fd = fd;
 609
 610         return 0;
 611 }
 612
 613 int ustcomm_disconnect(struct ustcomm_connection *conn)
 614 {
 615         return close(conn->fd);
 616 }
 617
 618 /* Open a connection to a traceable app.
 619  *
 620  * Return value:
 621  *  0: success
 622  * -1: error
 623  */
 624
 625 int ustcomm_connect_app(pid_t pid, struct ustcomm_connection *conn)
 626 {
 627         int result;
 628         char path[UNIX_PATH_MAX];
 629
 630
 631         result = snprintf(path, UNIX_PATH_MAX, "%s/%d", SOCK_DIR, pid);
 632         if(result >= UNIX_PATH_MAX) {
 633                 ERR("string overflow allocating socket name");
 634                 return -1;
 635         }
 636
 637         return ustcomm_connect_path(path, conn, pid);
 638 }
 639
 640 /* Close a connection to a traceable app. It frees the
 641  * resources. It however does not free the
 642  * ustcomm_connection itself.
 643  */
 644
 645 int ustcomm_close_app(struct ustcomm_connection *conn)
 646 {
 647         close(conn->fd);
 648         free(conn->recv_buf);
 649
 650         return 0;
 651 }
 652
 653 static int ensure_dir_exists(const char *dir)
 654 {
 655         struct stat st;
 656         int result;
 657
 658         if(!strcmp(dir, ""))
 659                 return -1;
 660
 661         result = stat(dir, &st);
 662         if(result == -1 && errno != ENOENT) {
 663                 return -1;
 664         }
 665         else if(result == -1) {
 666                 /* ENOENT */
 667                 int result;
 668
 669                 result = mkdir_p(dir, 0777);
 670                 if(result != 0) {
 671                         ERR("executing in recursive creation of directory %s", dir);
 672                         return -1;
 673                 }
 674         }
 675
 676         return 0;
 677 }
 678
 679 /* Called by an application to initialize its server so daemons can
 680  * connect to it.
 681  */
 682
 683 int ustcomm_init_app(pid_t pid, struct ustcomm_app *handle)
 684 {
 685         int result;
 686         char *name;
 687
 688         result = asprintf(&name, "%s/%d", SOCK_DIR, (int)pid);
 689         if(result >= UNIX_PATH_MAX) {
 690                 ERR("string overflow allocating socket name");
 691                 return -1;
 692         }
 693
 694         result = ensure_dir_exists(SOCK_DIR);
 695         if(result == -1) {
 696                 ERR("Unable to create socket directory %s", SOCK_DIR);
 697                 return -1;
 698         }
 699
 700         handle->server.listen_fd = init_named_socket(name, &(handle->server.socketpath));
 701         if(handle->server.listen_fd < 0) {
 702                 ERR("Error initializing named socket (%s). Check that directory exists and that it is writable.", name);
 703                 goto free_name;
 704         }
 705         free(name);
 706
 707         INIT_LIST_HEAD(&handle->server.connections);
 708
 709         return 0;
 710
 711 free_name:
 712         free(name);
 713         return -1;
 714 }
 715
 716 /* Used by the daemon to initialize its server so applications
 717  * can connect to it.
 718  */
 719
 720 int ustcomm_init_ustd(struct ustcomm_ustd *handle, const char *sock_path)
 721 {
 722         char *name;
 723         int retval = 0;
 724
 725         if(sock_path) {
 726                 asprintf(&name, "%s", sock_path);
 727         }
 728         else {
 729                 int result;
 730
 731                 /* Only check if socket dir exists if we are using the default directory */
 732                 result = ensure_dir_exists(SOCK_DIR);
 733                 if(result == -1) {
 734                         ERR("Unable to create socket directory %s", SOCK_DIR);
 735                         return -1;
 736                 }
 737
 738                 asprintf(&name, "%s/%s", SOCK_DIR, "ustd");
 739         }
 740
 741         handle->server.listen_fd = init_named_socket(name, &handle->server.socketpath);
 742         if(handle->server.listen_fd < 0) {
 743                 ERR("error initializing named socket at %s", name);
 744                 retval = -1;
 745                 goto free_name;
 746         }
 747
 748         INIT_LIST_HEAD(&handle->server.connections);
 749
 750 free_name:
 751         free(name);
 752
 753         return retval;
 754 }
 755
 756 static void ustcomm_fini_server(struct ustcomm_server *server, int keep_socket_file)
 757 {
 758         int result;
 759         struct stat st;
 760
 761         if(!keep_socket_file) {
 762                 /* Destroy socket */
 763                 result = stat(server->socketpath, &st);
 764                 if(result == -1) {
 765                         PERROR("stat (%s)", server->socketpath);
 766                         return;
 767                 }
 768
 769                 /* Paranoid check before deleting. */
 770                 result = S_ISSOCK(st.st_mode);
 771                 if(!result) {
 772                         ERR("The socket we are about to delete is not a socket.");
 773                         return;
 774                 }
 775
 776                 result = unlink(server->socketpath);
 777                 if(result == -1) {
 778                         PERROR("unlink");
 779                 }
 780         }
 781
 782         free(server->socketpath);
 783
 784         result = close(server->listen_fd);
 785         if(result == -1) {
 786                 PERROR("close");
 787                 return;
 788         }
 789 }
 790
 791 /* Free a traceable application server */
 792
 793 void ustcomm_fini_app(struct ustcomm_app *handle, int keep_socket_file)
 794 {
 795         ustcomm_fini_server(&handle->server, keep_socket_file);
 796 }
 797
 798 /* Free a ustd server */
 799
 800 void ustcomm_fini_ustd(struct ustcomm_ustd *handle)
 801 {
 802         ustcomm_fini_server(&handle->server, 0);
 803 }
 804
 805 static const char *find_tok(const char *str)
 806 {
 807         while(*str == ' ') {
 808                 str++;
 809
 810                 if(*str == 0)
 811                         return NULL;
 812         }
 813
 814         return str;
 815 }
 816
 817 static const char *find_sep(const char *str)
 818 {
 819         while(*str != ' ') {
 820                 str++;
 821
 822                 if(*str == 0)
 823                         break;
 824         }
 825
 826         return str;
 827 }
 828
 829 int nth_token_is(const char *str, const char *token, int tok_no)
 830 {
 831         int i;
 832         const char *start;
 833         const char *end;
 834
 835         for(i=0; i<=tok_no; i++) {
 836                 str = find_tok(str);
 837                 if(str == NULL)
 838                         return -1;
 839
 840                 start = str;
 841
 842                 str = find_sep(str);
 843                 if(str == NULL)
 844                         return -1;
 845
 846                 end = str;
 847         }
 848
 849         if(end-start != strlen(token))
 850                 return 0;
 851
 852         if(strncmp(start, token, end-start))
 853                 return 0;
 854
 855         return 1;
 856 }
 857
 858 char *nth_token(const char *str, int tok_no)
 859 {
 860         static char *retval = NULL;
 861         int i;
 862         const char *start;
 863         const char *end;
 864
 865         for(i=0; i<=tok_no; i++) {
 866                 str = find_tok(str);
 867                 if(str == NULL)
 868                         return NULL;
 869
 870                 start = str;
 871
 872                 str = find_sep(str);
 873                 if(str == NULL)
 874                         return NULL;
 875
 876                 end = str;
 877         }
 878
 879         if(retval) {
 880                 free(retval);
 881                 retval = NULL;
 882         }
 883
 884         asprintf(&retval, "%.*s", (int)(end-start), start);
 885
 886         return retval;
 887 }
 888
 889 /* Callback from multipoll.
 890  * Receive a new connection on the listening socket.
 891  */
 892
 893 static int process_mp_incoming_conn(void *priv, int fd, short events)
 894 {
 895         struct ustcomm_connection *newconn;
 896         struct ustcomm_server *server = (struct ustcomm_server *) priv;
 897         int newfd;
 898         int result;
 899
 900         result = newfd = accept(server->listen_fd, NULL, NULL);
 901         if(result == -1) {
 902                 PERROR("accept");
 903                 return -1;
 904         }
 905
 906         newconn = (struct ustcomm_connection *) malloc(sizeof(struct ustcomm_connection));
 907         if(newconn == NULL) {
 908                 ERR("malloc returned NULL");
 909                 return -1;
 910         }
 911
 912         ustcomm_init_connection(newconn);
 913         newconn->fd = newfd;
 914
 915         list_add(&newconn->list, &server->connections);
 916
 917         return 0;
 918 }
 919
 920 /* Callback from multipoll.
 921  * Receive a message on an existing connection.
 922  */
 923
 924 static int process_mp_conn_msg(void *priv, int fd, short revents)
 925 {
 926         struct ustcomm_multipoll_conn_info *mpinfo = (struct ustcomm_multipoll_conn_info *) priv;
 927         int result;
 928         char *msg;
 929         struct ustcomm_source src;
 930
 931         if(revents) {
 932                 src.fd = fd;
 933
 934                 result = recv_message_conn(mpinfo->conn, &msg);
 935                 if(result == -1) {
 936                         ERR("error in recv_message_conn");
 937                 }
 938
 939                 else if(result == 0) {
 940                         /* connection finished */
 941                         ustcomm_close_app(mpinfo->conn);
 942                         list_del(&mpinfo->conn->list);
 943                         free(mpinfo->conn);
 944                 }
 945                 else {
 946                         mpinfo->cb(msg, &src);
 947                         free(msg);
 948                 }
 949         }
 950
 951         return 0;
 952 }
 953
 954 int free_ustcomm_client_poll(void *data)
 955 {
 956         free(data);
 957         return 0;
 958 }
 959
 960 void ustcomm_mp_add_app_clients(struct mpentries *ent, struct ustcomm_app *app, int (*cb)(struct ustcomm_connection *conn, char *msg))
 961 {
 962         struct ustcomm_connection *conn;
 963
 964         /* add listener socket */
 965         multipoll_add(ent, app->server.listen_fd, POLLIN, process_mp_incoming_conn, &app->server, NULL);
 966
 967         list_for_each_entry(conn, &app->server.connections, list) {
 968                 struct ustcomm_multipoll_conn_info *mpinfo = (struct ustcomm_multipoll_conn_info *) malloc(sizeof(struct ustcomm_multipoll_conn_info));
 969                 mpinfo->conn = conn;
 970                 mpinfo->cb = cb;
 971                 multipoll_add(ent, conn->fd, POLLIN, process_mp_conn_msg, mpinfo, free_ustcomm_client_poll);
 972         }
 973 }