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