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