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