src/utils/socket_bundle.c

   1 #include "socket_bundle.h"
   2
   3 /* --------------------------------------------------------------------
   4         Test Code
   5         -------------------------------------------------------------------- */
   6 /*
   7 int count = 0;
   8 void printme(void* blob, socket_manager* mgr,
   9                 int sock_fd, char* data, int parent_id) {
  10
  11         fprintf(stderr, "Got data from socket %d with parent %d => %s",
  12                         sock_fd, parent_id, data );
  13
  14         socket_send(sock_fd, data);
  15
  16         if(count++ > 2) {
  17                 socket_disconnect(mgr, sock_fd);
  18                 _socket_print_list(mgr);
  19         }
  20 }
  21
  22 int main(int argc, char* argv[]) {
  23         socket_manager manager;
  24         memset(&manager, 0, sizeof(socket_manager));
  25         int port = 11000;
  26         if(argv[1])
  27                 port = atoi(argv[1]);
  28
  29         manager.data_received = &printme;
  30         socket_open_tcp_server(&manager, port);
  31
  32         while(1)
  33                 socket_wait_all(&manager, -1);
  34
  35         return 0;
  36 }
  37 */
  38 /* -------------------------------------------------------------------- */
  39
  40
  41
  42 socket_node* _socket_add_node(socket_manager* mgr,
  43                 int endpoint, int addr_type, int sock_fd, int parent_id ) {
  44
  45         if(mgr == NULL) return NULL;
  46         osrfLogInternal( OSRF_LOG_MARK, "Adding socket node with fd %d", sock_fd);
  47         socket_node* new_node = safe_malloc(sizeof(socket_node));
  48
  49         new_node->endpoint      = endpoint;
  50         new_node->addr_type     = addr_type;
  51         new_node->sock_fd               = sock_fd;
  52         new_node->next                  = NULL;
  53         new_node->parent_id = 0;
  54         if(parent_id > 0)
  55                 new_node->parent_id = parent_id;
  56
  57         new_node->next                  = mgr->socket;
  58         mgr->socket                             = new_node;
  59         return new_node;
  60 }
  61
  62 /* creates a new server socket node and adds it to the socket set.
  63         returns new socket fd on success.  -1 on failure.
  64         socket_type is one of INET or UNIX  */
  65 int socket_open_tcp_server(socket_manager* mgr, int port, char* listen_ip) {
  66
  67         if( mgr == NULL ) {
  68                 osrfLogWarning( OSRF_LOG_MARK, "socket_open_tcp_server(): NULL mgr");
  69                 return -1;
  70         }
  71
  72         int sock_fd;
  73         struct sockaddr_in server_addr;
  74
  75         sock_fd = socket(AF_INET, SOCK_STREAM, 0);
  76
  77         if(sock_fd < 0) {
  78                 osrfLogWarning( OSRF_LOG_MARK, "tcp_server_connect(): Unable to create socket");
  79                 return -1;
  80         }
  81
  82         server_addr.sin_family = AF_INET;
  83
  84         if(listen_ip != NULL) {
  85                 server_addr.sin_addr.s_addr = inet_addr(listen_ip);
  86         } else {
  87                 server_addr.sin_addr.s_addr = htonl(INADDR_ANY);
  88         }
  89
  90         server_addr.sin_port = htons(port);
  91
  92         if(bind( sock_fd, (struct sockaddr*) &server_addr, sizeof(server_addr)) < 0) {
  93                 osrfLogWarning( OSRF_LOG_MARK, "tcp_server_connect(): cannot bind to port %d", port );
  94                 return -1;
  95         }
  96
  97         if(listen(sock_fd, 20) == -1) {
  98                 osrfLogWarning( OSRF_LOG_MARK, "tcp_server_connect(): listen() returned error");
  99                 return -1;
 100         }
 101
 102         _socket_add_node(mgr, SERVER_SOCKET, INET, sock_fd, 0);
 103         return sock_fd;
 104 }
 105
 106 int socket_open_unix_server(socket_manager* mgr, char* path) {
 107         if(mgr == NULL || path == NULL) return -1;
 108
 109         osrfLogDebug( OSRF_LOG_MARK, "opening unix socket at %s", path);
 110         int sock_fd;
 111         struct sockaddr_un server_addr;
 112
 113         sock_fd = socket(AF_UNIX, SOCK_STREAM, 0);
 114         if(sock_fd < 0){
 115                 osrfLogWarning( OSRF_LOG_MARK, "socket_open_unix_server(): socket() failed");
 116                 return -1;
 117         }
 118
 119         server_addr.sun_family = AF_UNIX;
 120         strcpy(server_addr.sun_path, path);
 121
 122         if( bind(sock_fd, (struct sockaddr*) &server_addr,
 123                                 sizeof(struct sockaddr_un)) < 0) {
 124                 osrfLogWarning( OSRF_LOG_MARK,
 125                         "socket_open_unix_server(): cannot bind to unix port %s", path );
 126                 return -1;
 127         }
 128
 129         if(listen(sock_fd, 20) == -1) {
 130                 osrfLogWarning( OSRF_LOG_MARK, "socket_open_unix_server(): listen() returned error");
 131                 return -1;
 132         }
 133
 134         osrfLogDebug( OSRF_LOG_MARK, "unix socket successfully opened");
 135
 136         int i = 1;
 137
 138         /* causing problems with router for some reason ... */
 139         //osrfLogDebug( OSRF_LOG_MARK, "Setting SO_REUSEADDR");
 140         //setsockopt(sock_fd, SOL_SOCKET, SO_REUSEADDR, &i, sizeof(i));
 141
 142         //osrfLogDebug( OSRF_LOG_MARK, "Setting TCP_NODELAY");
 143         setsockopt(sock_fd, IPPROTO_TCP, TCP_NODELAY, &i, sizeof(i));
 144
 145         _socket_add_node(mgr, SERVER_SOCKET, UNIX, sock_fd, 0);
 146         return sock_fd;
 147 }
 148
 149
 150
 151 int socket_open_udp_server(
 152                 socket_manager* mgr, int port, char* listen_ip ) {
 153
 154         int sockfd;
 155         struct sockaddr_in server_addr;
 156
 157         if( (sockfd = socket( AF_INET, SOCK_DGRAM, 0 )) < 0 ) {
 158                 osrfLogWarning( OSRF_LOG_MARK, "Unable to create UDP socket");
 159                 return -1;
 160         }
 161
 162         server_addr.sin_family = AF_INET;
 163         server_addr.sin_port = htons(port);
 164         if(listen_ip) server_addr.sin_addr.s_addr = inet_addr(listen_ip);
 165         else server_addr.sin_addr.s_addr = htonl(INADDR_ANY);
 166
 167         if( (bind (sockfd, (struct sockaddr *) &server_addr,sizeof(server_addr))) ) {
 168                 osrfLogWarning( OSRF_LOG_MARK, "Unable to bind to UDP port %d", port);
 169                 return -1;
 170         }
 171
 172         _socket_add_node(mgr, SERVER_SOCKET, INET, sockfd, 0);
 173         return sockfd;
 174 }
 175
 176
 177 int socket_open_tcp_client(socket_manager* mgr, int port, char* dest_addr) {
 178
 179         struct sockaddr_in remoteAddr, localAddr;
 180    struct hostent *hptr;
 181    int sock_fd;
 182
 183    // ------------------------------------------------------------------
 184    // Create the socket
 185    // ------------------------------------------------------------------
 186    if( (sock_fd = socket( AF_INET, SOCK_STREAM, 0 )) < 0 ) {
 187       osrfLogWarning( OSRF_LOG_MARK,  "tcp_connect(): Cannot create socket" );
 188       return -1;
 189    }
 190
 191         int i = 1;
 192         //osrfLogDebug( OSRF_LOG_MARK, "Setting TCP_NODELAY");
 193         setsockopt(sock_fd, IPPROTO_TCP, TCP_NODELAY, &i, sizeof(i));
 194
 195
 196    // ------------------------------------------------------------------
 197    // Get the hostname
 198    // ------------------------------------------------------------------
 199    if( (hptr = gethostbyname( dest_addr ) ) == NULL ) {
 200       osrfLogWarning(  OSRF_LOG_MARK, "tcp_connect(): Unknown Host => %s", dest_addr );
 201       return -1;
 202    }
 203
 204    // ------------------------------------------------------------------
 205    // Construct server info struct
 206    // ------------------------------------------------------------------
 207    memset( &remoteAddr, 0, sizeof(remoteAddr));
 208    remoteAddr.sin_family = AF_INET;
 209    remoteAddr.sin_port = htons( port );
 210    memcpy( (char*) &remoteAddr.sin_addr.s_addr,
 211          hptr->h_addr_list[0], hptr->h_length );
 212
 213    // ------------------------------------------------------------------
 214    // Construct local info struct
 215    // ------------------------------------------------------------------
 216    memset( &localAddr, 0, sizeof( localAddr ) );
 217    localAddr.sin_family = AF_INET;
 218    localAddr.sin_addr.s_addr = htonl( INADDR_ANY );
 219    localAddr.sin_port = htons(0);
 220
 221    // ------------------------------------------------------------------
 222    // Bind to a local port
 223    // ------------------------------------------------------------------
 224    if( bind( sock_fd, (struct sockaddr *) &localAddr, sizeof( localAddr ) ) < 0 ) {
 225       osrfLogWarning(  OSRF_LOG_MARK, "tcp_connect(): Cannot bind to local port" );
 226       return -1;
 227    }
 228
 229    // ------------------------------------------------------------------
 230    // Connect to server
 231    // ------------------------------------------------------------------
 232    if( connect( sock_fd, (struct sockaddr*) &remoteAddr, sizeof( struct sockaddr_in ) ) < 0 ) {
 233       osrfLogWarning(  OSRF_LOG_MARK, "tcp_connect(): Cannot connect to server %s", dest_addr );
 234       return -1;
 235    }
 236
 237         _socket_add_node(mgr, CLIENT_SOCKET, INET, sock_fd, -1 );
 238
 239    return sock_fd;
 240 }
 241
 242
 243 int socket_open_udp_client(
 244                 socket_manager* mgr, int port, char* dest_addr) {
 245
 246         int sockfd;
 247         struct sockaddr_in client_addr, server_addr;
 248         struct hostent* host;
 249
 250         if( (host = gethostbyname(dest_addr)) == NULL) {
 251                 osrfLogWarning( OSRF_LOG_MARK, "Unable to resolve host: %s", dest_addr);
 252                 return -1;
 253         }
 254
 255         server_addr.sin_family = host->h_addrtype;
 256         memcpy((char *) &server_addr.sin_addr.s_addr,
 257                              host->h_addr_list[0], host->h_length);
 258         server_addr.sin_port = htons(port);
 259
 260         if( (sockfd = socket(AF_INET,SOCK_DGRAM,0)) < 0 ) {
 261                 osrfLogWarning( OSRF_LOG_MARK, "Unable to create UDP socket");
 262                 return -1;
 263         }
 264
 265         client_addr.sin_family = AF_INET;
 266         client_addr.sin_addr.s_addr = htonl(INADDR_ANY);
 267         client_addr.sin_port = htons(0);
 268
 269         if( (bind(sockfd, (struct sockaddr *) &client_addr, sizeof(client_addr))) < 0 ) {
 270                 osrfLogWarning( OSRF_LOG_MARK, "Unable to bind UDP socket");
 271                 return -1;
 272         }
 273
 274         _socket_add_node(mgr, CLIENT_SOCKET, INET, sockfd, -1 );
 275
 276         return sockfd;
 277 }
 278
 279
 280 int socket_open_unix_client(socket_manager* mgr, char* sock_path) {
 281
 282         int sock_fd, len;
 283    struct sockaddr_un usock;
 284
 285    if( (sock_fd = socket( AF_UNIX, SOCK_STREAM, 0 )) < 0 ) {
 286                 osrfLogWarning(  OSRF_LOG_MARK, "Cannot create socket" );
 287                 return -1;
 288         }
 289
 290    usock.sun_family = AF_UNIX;
 291    strcpy( usock.sun_path, sock_path );
 292
 293    len = sizeof( usock.sun_family ) + strlen( usock.sun_path );
 294
 295    if( connect( sock_fd, (struct sockaddr *) &usock, len ) < 0 ) {
 296       osrfLogWarning(  OSRF_LOG_MARK, "Error connecting to unix socket" );
 297                 return -1;
 298         }
 299
 300         _socket_add_node(mgr, CLIENT_SOCKET, UNIX, sock_fd, -1 );
 301
 302    return sock_fd;
 303 }
 304
 305
 306
 307 /* returns the socket_node with the given sock_fd */
 308 socket_node* socket_find_node(socket_manager* mgr, int sock_fd) {
 309         if(mgr == NULL) return NULL;
 310         socket_node* node = mgr->socket;
 311         while(node) {
 312                 if(node->sock_fd == sock_fd)
 313                         return node;
 314                 node = node->next;
 315         }
 316         return NULL;
 317 }
 318
 319 /* removes the node with the given sock_fd from the list and frees it */
 320 void socket_remove_node(socket_manager* mgr, int sock_fd) {
 321
 322         if(mgr == NULL) return;
 323
 324         osrfLogDebug( OSRF_LOG_MARK, "removing socket %d", sock_fd);
 325
 326         socket_node* head = mgr->socket;
 327         socket_node* tail = head;
 328         if(head == NULL) return;
 329
 330         /* if removing the first node in the list */
 331         if(head->sock_fd == sock_fd) {
 332                 mgr->socket = head->next;
 333                 free(head);
 334                 return;
 335         }
 336
 337         head = head->next;
 338
 339         /* if removing any other node */
 340         while(head) {
 341                 if(head->sock_fd == sock_fd) {
 342                         tail->next = head->next;
 343                         free(head);
 344                         return;
 345                 }
 346                 tail = head;
 347                 head = head->next;
 348         }
 349 }
 350
 351
 352
 353 void _socket_print_list(socket_manager* mgr) {
 354         if(mgr == NULL) return;
 355         socket_node* node = mgr->socket;
 356         osrfLogDebug( OSRF_LOG_MARK, "socket_node list: [");
 357         while(node) {
 358                 osrfLogDebug( OSRF_LOG_MARK, "sock_fd: %d | parent_id: %d",
 359                                 node->sock_fd, node->parent_id);
 360                 node = node->next;
 361         }
 362         osrfLogDebug( OSRF_LOG_MARK, "]");
 363 }
 364
 365 /* sends the given data to the given socket */
 366 int socket_send(int sock_fd, const char* data) {
 367         return _socket_send( sock_fd, data, 0);
 368 }
 369
 370
 371 int _socket_send(int sock_fd, const char* data, int flags) {
 372
 373         signal(SIGPIPE, SIG_IGN); /* in case a unix socket was closed */
 374
 375         size_t r = send( sock_fd, data, strlen(data), flags );
 376
 377         if( r == -1 ) {
 378                 osrfLogWarning( OSRF_LOG_MARK, "tcp_server_send(): Error sending data with return %d", r );
 379                 osrfLogWarning( OSRF_LOG_MARK, "Last Sys Error: %s", strerror(errno));
 380                 return -1;
 381         }
 382
 383         return 0;
 384 }
 385
 386
 387 int socket_send_nowait( int sock_fd, const char* data) {
 388         return _socket_send( sock_fd, data, MSG_DONTWAIT);
 389 }
 390
 391
 392 /*
 393  * Waits at most usecs microseconds for the send buffer of the given
 394  * socket to accept new data.  This does not guarantee that the
 395  * socket will accept all the data we want to give it.
 396  */
 397 int socket_send_timeout( int sock_fd, const char* data, int usecs ) {
 398
 399         fd_set write_set;
 400         FD_ZERO( &write_set );
 401         FD_SET( sock_fd, &write_set );
 402
 403         int mil = 1000000;
 404         int secs = (int) usecs / mil;
 405         usecs = usecs - (secs * mil);
 406
 407         struct timeval tv;
 408         tv.tv_sec = secs;
 409         tv.tv_usec = usecs;
 410
 411         osrfLogInfo(OSRF_LOG_MARK, "Socket waiting on select before send");
 412         int ret = select( sock_fd + 1, NULL, &write_set, NULL, &tv);
 413         osrfLogInfo(OSRF_LOG_MARK, "Socket done waiting");
 414
 415         if( ret > 0 ) return _socket_send( sock_fd, data, 0);
 416
 417         osrfLogError(OSRF_LOG_MARK, "socket_send_timeout(): "
 418                 "timed out on send for socket %d after %d secs, %d usecs", sock_fd, secs, usecs );
 419
 420         return -1;
 421 }
 422
 423
 424 /* disconnects the node with the given sock_fd and removes
 425         it from the socket set */
 426 void socket_disconnect(socket_manager* mgr, int sock_fd) {
 427         osrfLogInternal( OSRF_LOG_MARK, "Closing socket %d", sock_fd);
 428         close( sock_fd );
 429         socket_remove_node(mgr, sock_fd);
 430 }
 431
 432
 433 /* we assume that if select() fails, the socket is no longer valid */
 434 int socket_connected(int sock_fd) {
 435         fd_set read_set;
 436         FD_ZERO( &read_set );
 437         FD_SET( sock_fd, &read_set );
 438         if( select( sock_fd + 1, &read_set, NULL, NULL, NULL) == -1 )
 439                 return 0;
 440         return 1;
 441
 442 }
 443
 444 /* this only waits on the server socket and does not handle the actual
 445         data coming in from the client..... XXX */
 446 int socket_wait(socket_manager* mgr, int timeout, int sock_fd) {
 447
 448         int retval = 0;
 449         fd_set read_set;
 450         FD_ZERO( &read_set );
 451         FD_SET( sock_fd, &read_set );
 452
 453         struct timeval tv;
 454         tv.tv_sec = timeout;
 455         tv.tv_usec = 0;
 456
 457         if( timeout < 0 ) {
 458
 459                 // If timeout is -1, we block indefinitely
 460                 if( (retval = select( sock_fd + 1, &read_set, NULL, NULL, NULL)) == -1 ) {
 461                         osrfLogWarning( OSRF_LOG_MARK, "Sys Error: %s", strerror(errno));
 462                         osrfLogWarning( OSRF_LOG_MARK, "Call to select interrupted");
 463                         return -1;
 464                 }
 465
 466         } else if( timeout > 0 ) { /* timeout of 0 means don't block */
 467
 468                 if( (retval = select( sock_fd + 1, &read_set, NULL, NULL, &tv)) == -1 ) {
 469                         osrfLogWarning( OSRF_LOG_MARK, "Sys Error: %s", strerror(errno));
 470                         osrfLogWarning( OSRF_LOG_MARK,  "Call to select interrupted" );
 471                         return -1;
 472                 }
 473         }
 474
 475         osrfLogInternal( OSRF_LOG_MARK, "%d active sockets after select()", retval);
 476         return _socket_route_data_id(mgr, sock_fd);
 477 }
 478
 479
 480 int socket_wait_all(socket_manager* mgr, int timeout) {
 481
 482         if(mgr == NULL) {
 483                 osrfLogWarning( OSRF_LOG_MARK,  "tcp_wait(): null mgr" );
 484                 return -1;
 485         }
 486
 487         int retval = 0;
 488         fd_set read_set;
 489         FD_ZERO( &read_set );
 490
 491         socket_node* node = mgr->socket;
 492         int max_fd = 0;
 493         while(node) {
 494                 osrfLogInternal( OSRF_LOG_MARK, "Adding socket fd %d to select set",node->sock_fd);
 495                 FD_SET( node->sock_fd, &read_set );
 496                 if(node->sock_fd > max_fd) max_fd = node->sock_fd;
 497                 node = node->next;
 498         }
 499         max_fd += 1;
 500
 501         struct timeval tv;
 502         tv.tv_sec = timeout;
 503         tv.tv_usec = 0;
 504
 505         if( timeout < 0 ) {
 506
 507                 // If timeout is -1, there is no timeout passed to the call to select
 508                 if( (retval = select( max_fd, &read_set, NULL, NULL, NULL)) == -1 ) {
 509                         osrfLogWarning( OSRF_LOG_MARK, "Call to select interrupted (returned -1)");
 510                         osrfLogWarning( OSRF_LOG_MARK, "Sys Error: %s", strerror(errno));
 511                         return -1;
 512                 }
 513
 514         } else if( timeout != 0 ) { /* timeout of 0 means don't block */
 515
 516                 if( (retval = select( max_fd, &read_set, NULL, NULL, &tv)) == -1 ) {
 517                         osrfLogWarning( OSRF_LOG_MARK,  "Call to select interrupted (returned -1)" );
 518                         osrfLogWarning( OSRF_LOG_MARK, "Sys Error: %s", strerror(errno));
 519                         return -1;
 520                 }
 521         }
 522
 523         osrfLogDebug( OSRF_LOG_MARK, "%d active sockets after select()", retval);
 524         return _socket_route_data(mgr, retval, &read_set);
 525 }
 526
 527 /* determines if we'er receiving a new client or data
 528         on an existing client */
 529 int _socket_route_data(
 530         socket_manager* mgr, int num_active, fd_set* read_set) {
 531
 532         if(!(mgr && read_set)) return -1;
 533
 534         int last_failed_id = -1;
 535
 536
 537         /* come back here if someone yanks a socket_node from beneath us */
 538         while(1) {
 539
 540                 socket_node* node = mgr->socket;
 541                 int handled = 0;
 542                 int status = 0;
 543
 544                 while(node && (handled < num_active)) {
 545
 546                         int sock_fd = node->sock_fd;
 547
 548                         if(last_failed_id != -1) {
 549                                 /* in case it was not removed by our overlords */
 550                                 osrfLogInternal( OSRF_LOG_MARK, "Attempting to remove last_failed_id of %d", last_failed_id);
 551                                 socket_remove_node( mgr, last_failed_id );
 552                                 last_failed_id = -1;
 553                                 status = -1;
 554                                 break;
 555                         }
 556
 557                         /* does this socket have data? */
 558                         if( FD_ISSET( sock_fd, read_set ) ) {
 559
 560                                 osrfLogInternal( OSRF_LOG_MARK, "Socket %d active", sock_fd);
 561                                 handled++;
 562                                 FD_CLR(sock_fd, read_set);
 563
 564                                 if(node->endpoint == SERVER_SOCKET)
 565                                         _socket_handle_new_client(mgr, node);
 566
 567                                 else
 568                                         status = _socket_handle_client_data(mgr, node);
 569
 570                                 /* someone may have yanked a socket_node out from under
 571                                         us...start over with the first socket */
 572                                 if(status == -1)  {
 573                                         last_failed_id = sock_fd;
 574                                         osrfLogInternal( OSRF_LOG_MARK, "Backtracking back to start of loop because "
 575                                                         "of -1 return code from _socket_handle_client_data()");
 576                                 }
 577                         }
 578
 579                         if(status == -1) break;
 580                         node = node->next;
 581
 582                 } // is_set
 583
 584                 if(status == 0) break;
 585                 if(status == -1) status = 0;
 586         }
 587
 588         return 0;
 589 }
 590
 591
 592 int _socket_route_data_id( socket_manager* mgr, int sock_id) {
 593         socket_node* node = socket_find_node(mgr, sock_id);
 594         int status = 0;
 595
 596         if(node) {
 597                 if(node->endpoint == SERVER_SOCKET)
 598                         _socket_handle_new_client(mgr, node);
 599
 600                 if(node->endpoint == CLIENT_SOCKET )
 601                         status = _socket_handle_client_data(mgr, node);
 602
 603                 if(status == -1) socket_remove_node(mgr, sock_id);
 604                 return 0;
 605         }
 606
 607         return -1;
 608 }
 609
 610
 611 int _socket_handle_new_client(socket_manager* mgr, socket_node* node) {
 612         if(mgr == NULL || node == NULL) return -1;
 613
 614         int new_sock_fd;
 615         new_sock_fd = accept(node->sock_fd, NULL, NULL);
 616         if(new_sock_fd < 0) {
 617                 osrfLogWarning( OSRF_LOG_MARK, "_socket_route_data(): accept() failed");
 618                 return -1;
 619         }
 620
 621         if(node->addr_type == INET) {
 622                 _socket_add_node(mgr, CLIENT_SOCKET, INET, new_sock_fd, node->sock_fd);
 623                 osrfLogDebug( OSRF_LOG_MARK, "Adding new INET client for %d", node->sock_fd);
 624
 625         } else if(node->addr_type == UNIX) {
 626                 _socket_add_node(mgr, CLIENT_SOCKET, UNIX, new_sock_fd, node->sock_fd);
 627                 osrfLogDebug( OSRF_LOG_MARK, "Adding new UNIX client for %d", node->sock_fd);
 628         }
 629
 630         return 0;
 631 }
 632
 633
 634 int _socket_handle_client_data(socket_manager* mgr, socket_node* node) {
 635         if(mgr == NULL || node == NULL) return -1;
 636
 637         char buf[RBUFSIZE];
 638         int read_bytes;
 639         int sock_fd = node->sock_fd;
 640
 641         memset(buf, 0, RBUFSIZE);
 642         set_fl(sock_fd, O_NONBLOCK);
 643
 644         osrfLogInternal( OSRF_LOG_MARK, "%d : Received data at %lf\n", getpid(), get_timestamp_millis());
 645
 646         while( (read_bytes = recv(sock_fd, buf, RBUFSIZE-1, 0) ) > 0 ) {
 647                 osrfLogInternal( OSRF_LOG_MARK, "Socket %d Read %d bytes and data: %s", sock_fd, read_bytes, buf);
 648                 if(mgr->data_received)
 649                         mgr->data_received(mgr->blob, mgr, sock_fd, buf, node->parent_id);
 650
 651                 memset(buf, 0, RBUFSIZE);
 652         }
 653
 654         if(socket_find_node(mgr, sock_fd)) {  /* someone may have closed this socket */
 655                 clr_fl(sock_fd, O_NONBLOCK);
 656                 if(read_bytes < 0) {
 657                         if( errno != EAGAIN )
 658                                 osrfLogWarning( OSRF_LOG_MARK,  " * Error reading socket with errno %d", errno );
 659                 }
 660
 661         } else { return -1; } /* inform the caller that this node has been tampered with */
 662
 663         if(read_bytes == 0) {  /* socket closed by client */
 664                 if(mgr->on_socket_closed) {
 665                         mgr->on_socket_closed(mgr->blob, sock_fd);
 666                         return -1;
 667                 }
 668         }
 669
 670         return 0;
 671
 672 }
 673
 674
 675 void socket_manager_free(socket_manager* mgr) {
 676         if(mgr == NULL) return;
 677         socket_node* tmp;
 678         while(mgr->socket) {
 679                 tmp = mgr->socket->next;
 680                 socket_disconnect(mgr, mgr->socket->sock_fd);
 681                 mgr->socket = tmp;
 682         }
 683         free(mgr);
 684
 685 }