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