Eliminated _socket_route_data_id() as a separate function, incorporating
[OpenSRF.git] / src / libopensrf / socket_bundle.c
1 /**
2         @file socket_bundle.c
3         @brief Collection of socket-handling routines.
4 */
5
6 #include <opensrf/socket_bundle.h>
7
8 struct socket_node_struct {
9         int endpoint;           /* SERVER_SOCKET or CLIENT_SOCKET */
10         int addr_type;          /* INET or UNIX */
11         int sock_fd;
12         int parent_id;          /* if we're a new client for a server socket, 
13         this points to the server socket we spawned from */
14         struct socket_node_struct* next;
15 };
16
17 /* buffer used to read from the sockets */
18 #define RBUFSIZE 1024
19
20 static socket_node* _socket_add_node(socket_manager* mgr,
21                 int endpoint, int addr_type, int sock_fd, int parent_id );
22 static socket_node* socket_find_node(socket_manager* mgr, int sock_fd);
23 static void socket_remove_node(socket_manager*, int sock_fd);
24 static int _socket_send(int sock_fd, const char* data, int flags);
25 static int _socket_route_data(socket_manager* mgr, int num_active, fd_set* read_set);
26 static int _socket_handle_new_client(socket_manager* mgr, socket_node* node);
27 static int _socket_handle_client_data(socket_manager* mgr, socket_node* node);
28
29
30 /* -------------------------------------------------------------------- 
31         Test Code 
32         -------------------------------------------------------------------- */
33 /*
34 int count = 0;
35 void printme(void* blob, socket_manager* mgr, 
36                 int sock_fd, char* data, int parent_id) {
37
38         fprintf(stderr, "Got data from socket %d with parent %d => %s", 
39                         sock_fd, parent_id, data );
40
41         socket_send(sock_fd, data);
42
43         if(count++ > 2) {
44                 socket_disconnect(mgr, sock_fd);
45                 _socket_print_list(mgr);
46         }
47 }
48
49 int main(int argc, char* argv[]) {
50         socket_manager manager;
51         memset(&manager, 0, sizeof(socket_manager));
52         int port = 11000;
53         if(argv[1])
54                 port = atoi(argv[1]);
55
56         manager.data_received = &printme;
57         socket_open_tcp_server(&manager, port);
58
59         while(1)
60                 socket_wait_all(&manager, -1);
61
62         return 0;
63 }
64 */
65 /* -------------------------------------------------------------------- */
66
67
68 /**
69         @brief Create a new socket_node and add it to a socket_manager's list.
70         @param mgr Pointer to the socket_manager.
71         @param endpoint SERVER_SOCKET or CLIENT_SOCKET, denoting how the socket is to be used.
72         @param addr_type address type: INET or UNIX.
73         @param sock_fd sock_fd for the new socket_node.
74         @param parent_id parent_id for the new node.
75         @return Pointer to the new socket_node.
76
77         If @a parent_id is negative, the new socket_node receives a parent_id of 0.
78 */
79 static socket_node* _socket_add_node(socket_manager* mgr, 
80                 int endpoint, int addr_type, int sock_fd, int parent_id ) {
81
82         if(mgr == NULL) return NULL;
83         osrfLogInternal( OSRF_LOG_MARK, "Adding socket node with fd %d", sock_fd);
84         socket_node* new_node = safe_malloc(sizeof(socket_node));
85
86         new_node->endpoint      = endpoint;
87         new_node->addr_type     = addr_type;
88         new_node->sock_fd       = sock_fd;
89         new_node->next          = NULL;
90         new_node->parent_id = 0;
91         if(parent_id > 0)
92                 new_node->parent_id = parent_id;
93
94         new_node->next                  = mgr->socket;
95         mgr->socket                             = new_node;
96         return new_node;
97 }
98
99 /**
100         @brief Create an TCP INET listener socket and add it to a socket_manager's list.
101         @param mgr Pointer to the socket manager that will own the socket.
102         @param port The port number to bind to.
103         @param listen_ip The IP address to bind to; or, NULL for INADDR_ANY.
104         @return The socket fd if successful; otherwise -1.
105
106         Calls: socket(), bind(), and listen().  Creates a SERVER_SOCKET.
107 */
108 int socket_open_tcp_server(socket_manager* mgr, int port, const char* listen_ip) {
109
110         if( mgr == NULL ) {
111                 osrfLogWarning( OSRF_LOG_MARK, "socket_open_tcp_server(): NULL mgr"); 
112                 return -1;
113         }
114
115         int sock_fd;
116         struct sockaddr_in server_addr;
117
118         server_addr.sin_family = AF_INET;
119
120         if(listen_ip != NULL) {
121                 struct in_addr addr;
122                 if( inet_aton( listen_ip, &addr ) )
123                         server_addr.sin_addr.s_addr = addr.s_addr;
124                 else {
125                         osrfLogError( OSRF_LOG_MARK, "Listener address is invalid: %s", listen_ip );
126                         return -1;
127                 }
128         } else {
129                 server_addr.sin_addr.s_addr = htonl(INADDR_ANY);
130         }
131
132         server_addr.sin_port = htons(port);
133
134         errno = 0;
135         sock_fd = socket(AF_INET, SOCK_STREAM, 0);
136         if(sock_fd < 0) {
137                 osrfLogWarning( OSRF_LOG_MARK, "socket_open_tcp_server(): Unable to create TCP socket: %s",
138                         strerror( errno ) );
139                 return -1;
140         }
141
142         errno = 0;
143         if(bind( sock_fd, (struct sockaddr*) &server_addr, sizeof(server_addr)) < 0) {
144                 osrfLogWarning( OSRF_LOG_MARK, "socket_open_tcp_server(): cannot bind to port %d: %s",
145                         port, strerror( errno ) );
146                 close( sock_fd );
147                 return -1;
148         }
149
150         errno = 0;
151         if(listen(sock_fd, 20) == -1) {
152                 osrfLogWarning( OSRF_LOG_MARK, "socket_open_tcp_server(): listen() returned error: %s",
153                         strerror( errno ) );
154                 close( sock_fd );
155                 return -1;
156         }
157
158         _socket_add_node(mgr, SERVER_SOCKET, INET, sock_fd, 0);
159         return sock_fd;
160 }
161
162 /**
163         @brief Create a UNIX domain listener socket and add it to the socket_manager's list.
164         @param mgr Pointer to the socket_manager that will own the socket.
165         @param path Name of the socket within the file system.
166         @return The socket fd if successful; otherwise -1.
167
168         Calls: socket(), bind(), listen().  Creates a SERVER_SOCKET.
169
170         Applies socket option TCP_NODELAY in order to reduce latency.
171  */
172 int socket_open_unix_server(socket_manager* mgr, const char* path) {
173         if(mgr == NULL || path == NULL) return -1;
174
175         osrfLogDebug( OSRF_LOG_MARK, "opening unix socket at %s", path);
176         int sock_fd;
177         struct sockaddr_un server_addr;
178
179         if(strlen(path) > sizeof(server_addr.sun_path) - 1)
180         {
181                 osrfLogWarning( OSRF_LOG_MARK, "socket_open_unix_server(): path too long: %s",
182                         path );
183                 return -1;
184         }
185
186         errno = 0;
187         sock_fd = socket(AF_UNIX, SOCK_STREAM, 0);
188         if(sock_fd < 0){
189                 osrfLogWarning( OSRF_LOG_MARK, "socket_open_unix_server(): socket() failed: %s",
190                         strerror( errno ) );
191                 return -1;
192         }
193
194         server_addr.sun_family = AF_UNIX;
195         strcpy(server_addr.sun_path, path);
196
197         errno = 0;
198         if( bind(sock_fd, (struct sockaddr*) &server_addr, 
199                                 sizeof(struct sockaddr_un)) < 0) {
200                 osrfLogWarning( OSRF_LOG_MARK, 
201                         "socket_open_unix_server(): cannot bind to unix port %s: %s",
202                         path, strerror( errno ) );
203                 close( sock_fd );
204                 return -1;
205         }
206
207         errno = 0;
208         if(listen(sock_fd, 20) == -1) {
209                 osrfLogWarning( OSRF_LOG_MARK, "socket_open_unix_server(): listen() returned error: %s",
210                         strerror( errno ) );
211                 close( sock_fd );
212                 return -1;
213         }
214
215         osrfLogDebug( OSRF_LOG_MARK, "unix socket successfully opened");
216         
217         int i = 1;
218
219         /* causing problems with router for some reason ... */
220         //osrfLogDebug( OSRF_LOG_MARK, "Setting SO_REUSEADDR");
221         //setsockopt(sock_fd, SOL_SOCKET, SO_REUSEADDR, &i, sizeof(i));
222         
223         //osrfLogDebug( OSRF_LOG_MARK, "Setting TCP_NODELAY");
224         setsockopt(sock_fd, IPPROTO_TCP, TCP_NODELAY, &i, sizeof(i));
225
226         _socket_add_node(mgr, SERVER_SOCKET, UNIX, sock_fd, 0);
227         return sock_fd;
228 }
229
230
231 /**
232         @brief Create a UDP socket for a server, and add it to a socket_manager's list.
233         @param mgr Pointer to the socket_manager that will own the socket.
234         @param port The port number to bind to.
235         @param listen_ip The IP address to bind to, or NULL for INADDR_ANY.
236         @return The socket fd if successful; otherwise -1.
237
238         Calls: socket(), bind().  Creates a SERVER_SOCKET.
239 */
240 int socket_open_udp_server( 
241                 socket_manager* mgr, int port, const char* listen_ip ) {
242
243         int sockfd;
244         struct sockaddr_in server_addr;
245
246         server_addr.sin_family = AF_INET;
247         server_addr.sin_port = htons(port);
248         if(listen_ip) {
249                 struct in_addr addr;
250                 if( inet_aton( listen_ip, &addr ) )
251                         server_addr.sin_addr.s_addr = addr.s_addr;
252                 else {
253                         osrfLogError( OSRF_LOG_MARK, "UDP listener address is invalid: %s", listen_ip );
254                         return -1;
255                 }
256         } else
257                 server_addr.sin_addr.s_addr = htonl(INADDR_ANY);
258
259         errno = 0;
260         if( (sockfd = socket( AF_INET, SOCK_DGRAM, 0 )) < 0 ) {
261                 osrfLogWarning( OSRF_LOG_MARK, "Unable to create UDP socket: %s", strerror( errno ) );
262                 return -1;
263         }
264
265         errno = 0;
266         if( (bind (sockfd, (struct sockaddr *) &server_addr,sizeof(server_addr))) ) {
267                 osrfLogWarning( OSRF_LOG_MARK, "Unable to bind to UDP port %d: %s",
268                         port, strerror( errno ) );
269                 close( sockfd );
270                 return -1;
271         }
272
273         _socket_add_node(mgr, SERVER_SOCKET, INET, sockfd, 0);
274         return sockfd;
275 }
276
277
278 /**
279         @brief Create a client TCP socket, connect with it, and add it to a socket_manager's list.
280         @param mgr Pointer to the socket_manager that will own the socket.
281         @param port What port number to connect to.
282         @param dest_addr Host name or IP address of the server to which we are connecting.
283         @return The socket fd if successful; otherwise -1.
284
285         Calls: getaddrinfo(), socket(), connect().  Creates a CLIENT_SOCKET.
286
287         Applies socket option TCP_NODELAY in order to reduce latency.
288  */
289 int socket_open_tcp_client(socket_manager* mgr, int port, const char* dest_addr) {
290
291         struct sockaddr_in remoteAddr;
292         int sock_fd;
293
294         // ------------------------------------------------------------------
295         // Get the IP address of the hostname (for TCP only)
296         // ------------------------------------------------------------------
297         struct addrinfo hints = { 0, 0, 0, 0, 0, NULL, NULL, NULL };
298         hints.ai_socktype = SOCK_STREAM;
299         struct addrinfo* addr_info = NULL;
300         errno = 0;
301         int rc = getaddrinfo( dest_addr, NULL, &hints, &addr_info );
302         if( rc || ! addr_info ) {
303                 osrfLogWarning( OSRF_LOG_MARK, "socket_open_tcp_client(): No Such Host => %s: %s",
304                         dest_addr, gai_strerror( rc ) );
305                 return -1;
306         }
307
308         // Look for an address supporting IPv4.  Someday we'll look for
309         // either IPv4 or IPv6, and branch according to what we find.
310         while( addr_info && addr_info->ai_family != PF_INET ) {
311                 addr_info = addr_info->ai_next;
312         }
313
314         if( ! addr_info ) {
315                 osrfLogWarning( OSRF_LOG_MARK,
316                         "socket_open_tcp_client(): Host %s does not support IPV4", dest_addr );
317                 return -1;      
318         }
319
320         // ------------------------------------------------------------------
321         // Create the socket
322         // ------------------------------------------------------------------
323         errno = 0;
324         if( (sock_fd = socket( AF_INET, SOCK_STREAM, 0 )) < 0 ) {
325                 osrfLogWarning( OSRF_LOG_MARK, "socket_open_tcp_client(): Cannot create TCP socket: %s",
326                         strerror( errno ) );
327                 return -1;
328         }
329
330         int i = 1;
331         setsockopt(sock_fd, IPPROTO_TCP, TCP_NODELAY, &i, sizeof(i));
332
333         // ------------------------------------------------------------------
334         // Construct server info struct
335         // ------------------------------------------------------------------
336         memset( &remoteAddr, 0, sizeof(remoteAddr));
337         remoteAddr.sin_family = AF_INET;
338         remoteAddr.sin_port = htons( port );
339         struct sockaddr_in* ai_addr_in = (struct sockaddr_in*) addr_info->ai_addr;
340         remoteAddr.sin_addr.s_addr = ai_addr_in->sin_addr.s_addr;
341
342         freeaddrinfo( addr_info );
343
344         // ------------------------------------------------------------------
345         // Connect to server
346         // ------------------------------------------------------------------
347         errno = 0;
348         if( connect( sock_fd, (struct sockaddr*) &remoteAddr, sizeof( struct sockaddr_in ) ) < 0 ) {
349                 osrfLogWarning( OSRF_LOG_MARK, "socket_open_tcp_client(): Cannot connect to server %s: %s",
350                         dest_addr, strerror(errno) );
351                 close( sock_fd );
352                 return -1;
353         }
354
355         _socket_add_node(mgr, CLIENT_SOCKET, INET, sock_fd, -1 );
356
357         return sock_fd;
358 }
359
360
361 /**
362         @brief Create a client UDP socket and add it to a socket_manager's list.
363         @param mgr Pointer to the socket_manager that will own the socket.
364         @return The socket fd if successful; otherwise -1.
365
366         Calls: socket().  Creates a CLIENT_SOCKET.
367 */
368 int socket_open_udp_client( socket_manager* mgr ) {
369
370         int sockfd;
371
372         errno = 0;
373         if( (sockfd = socket(AF_INET,SOCK_DGRAM,0)) < 0 ) {
374                 osrfLogWarning( OSRF_LOG_MARK,
375                         "socket_open_udp_client(): Unable to create UDP socket: %s", strerror( errno ) );
376                 return -1;
377         }
378
379         _socket_add_node(mgr, CLIENT_SOCKET, INET, sockfd, -1 );
380
381         return sockfd;
382 }
383
384
385 /**
386         @brief Create a UNIX domain client socket, connect with it, add it to the socket_manager's list
387         @param mgr Pointer to the socket_manager that will own the socket.
388         @param sock_path Name of the socket within the file system.
389         @return The socket fd if successful; otherwise -1.
390
391         Calls: socket(), connect().  Creates a CLIENT_SOCKET.
392 */
393 int socket_open_unix_client(socket_manager* mgr, const char* sock_path) {
394
395         int sock_fd, len;
396         struct sockaddr_un usock;
397
398         if(strlen(sock_path) > sizeof(usock.sun_path) - 1)
399         {
400                 osrfLogWarning( OSRF_LOG_MARK, "socket_open_unix_client(): path too long: %s",
401                    sock_path );
402                 return -1;
403         }
404
405         errno = 0;
406         if( (sock_fd = socket( AF_UNIX, SOCK_STREAM, 0 )) < 0 ) {
407                 osrfLogWarning(  OSRF_LOG_MARK, "socket_open_unix_client(): Cannot create UNIX socket: %s", strerror( errno ) );
408                 return -1;
409         }
410
411         usock.sun_family = AF_UNIX;
412         strcpy( usock.sun_path, sock_path );
413
414         len = sizeof( usock.sun_family ) + strlen( usock.sun_path );
415
416         errno = 0;
417         if( connect( sock_fd, (struct sockaddr *) &usock, len ) < 0 ) {
418                 osrfLogWarning(  OSRF_LOG_MARK, "Error connecting to unix socket: %s",
419                         strerror( errno ) );
420                 close( sock_fd );
421                 return -1;
422         }
423
424         _socket_add_node(mgr, CLIENT_SOCKET, UNIX, sock_fd, -1 );
425
426         return sock_fd;
427 }
428
429
430 /**
431         @brief Search a socket_manager's list for a socket node for a given file descriptor.
432         @param mgr Pointer to the socket manager.
433         @param sock_fd The file descriptor to be sought.
434         @return A pointer to the socket_node if found; otherwise NULL.
435
436         Traverse a linked list owned by the socket_manager.
437 */
438 static socket_node* socket_find_node(socket_manager* mgr, int sock_fd) {
439         if(mgr == NULL) return NULL;
440         socket_node* node = mgr->socket;
441         while(node) {
442                 if(node->sock_fd == sock_fd)
443                         return node;
444                 node = node->next;
445         }
446         return NULL;
447 }
448
449 /* removes the node with the given sock_fd from the list and frees it */
450 /**
451         @brief Remove a socket node for a given fd from a socket_manager's list.
452         @param mgr Pointer to the socket_manager.
453         @param sock_fd The file descriptor whose socket_node is to be removed.
454
455         This function does @em not close the socket.  It just removes a node from the list, and
456         frees it.  The disposition of the socket is the responsibility of the calling code.
457 */
458 static void socket_remove_node(socket_manager* mgr, int sock_fd) {
459
460         if(mgr == NULL) return;
461
462         osrfLogDebug( OSRF_LOG_MARK, "removing socket %d", sock_fd);
463
464         socket_node* head = mgr->socket;
465         socket_node* tail = head;
466         if(head == NULL) return;
467
468         /* if removing the first node in the list */
469         if(head->sock_fd == sock_fd) {
470                 mgr->socket = head->next;
471                 free(head);
472                 return;
473         }
474
475         head = head->next;
476
477         /* if removing any other node */
478         while(head) {
479                 if(head->sock_fd == sock_fd) {
480                         tail->next = head->next;
481                         free(head);
482                         return;
483                 }
484                 tail = head;
485                 head = head->next;
486         }
487 }
488
489
490 /**
491         @brief Write to the log: a list of socket_nodes in a socket_manager's list.
492         @param mgr Pointer to the socket_manager.
493
494         For testing and debugging.
495
496         The messages are issued as DEBG messages, and show each file descriptor and its parent.
497 */
498 void _socket_print_list(socket_manager* mgr) {
499         if(mgr == NULL) return;
500         socket_node* node = mgr->socket;
501         osrfLogDebug( OSRF_LOG_MARK, "socket_node list: [");
502         while(node) {
503                 osrfLogDebug( OSRF_LOG_MARK, "sock_fd: %d | parent_id: %d", 
504                                 node->sock_fd, node->parent_id);
505                 node = node->next;
506         }
507         osrfLogDebug( OSRF_LOG_MARK, "]");
508 }
509
510 /**
511         @brief Send a nul-terminated string over a socket.
512         @param sock_fd The file descriptor for the socket.
513         @param data Pointer to the string to be sent.
514         @return 0 if successful, -1 if not.
515
516         This function is a thin wrapper for _socket_send().
517 */
518 int socket_send(int sock_fd, const char* data) {
519         return _socket_send( sock_fd, data, 0);
520 }
521
522 /**
523         @brief Send a nul-terminated string over a socket.
524         @param sock_fd The file descriptor for the socket.
525         @param data Pointer to the string to be sent.
526         @param flags A set of bitflags to be passed to send().
527         @return 0 if successful, -1 if not.
528
529         This function is the final common pathway for all outgoing socket traffic.
530 */
531 static int _socket_send(int sock_fd, const char* data, int flags) {
532
533         signal(SIGPIPE, SIG_IGN); /* in case a unix socket was closed */
534
535         errno = 0;
536         size_t r = send( sock_fd, data, strlen(data), flags );
537         int local_errno = errno;
538
539         if( r == -1 ) {
540                 osrfLogWarning( OSRF_LOG_MARK, "_socket_send(): Error sending data with return %d", r );
541                 osrfLogWarning( OSRF_LOG_MARK, "Last Sys Error: %s", strerror(local_errno));
542                 return -1;
543         }
544
545         return 0;
546 }
547
548
549 /* sends the given data to the given socket. 
550  * sets the send flag MSG_DONTWAIT which will allow the 
551  * process to continue even if the socket buffer is full
552  * returns 0 on success, -1 otherwise */
553 //int socket_send_nowait( int sock_fd, const char* data) {
554 //      return _socket_send( sock_fd, data, MSG_DONTWAIT);
555 //}
556
557
558 /**
559         @brief Wait for a socket to be ready to send, and then send a string over it.
560         @param sock_fd File descriptor of the socket.
561         @param data Pointer to a nul-terminated string to be sent.
562         @param usecs How long to wait, in microseconds, before timing out.
563         @return 0 if successful, -1 if not.
564
565         The socket may not accept all the data we want to give it.
566 */
567 int socket_send_timeout( int sock_fd, const char* data, int usecs ) {
568
569         fd_set write_set;
570         FD_ZERO( &write_set );
571         FD_SET( sock_fd, &write_set );
572
573         const int mil = 1000000;
574         int secs = (int) usecs / mil;
575         usecs = usecs - (secs * mil);
576
577         struct timeval tv;
578         tv.tv_sec = secs;
579         tv.tv_usec = usecs;
580
581         errno = 0;
582         int ret = select( sock_fd + 1, NULL, &write_set, NULL, &tv);
583         if( ret > 0 ) return _socket_send( sock_fd, data, 0);
584
585         osrfLogError(OSRF_LOG_MARK, "socket_send_timeout(): "
586                 "timed out on send for socket %d after %d secs, %d usecs: %s",
587                 sock_fd, secs, usecs, strerror( errno ) );
588
589         return -1;
590 }
591
592
593 /* disconnects the node with the given sock_fd and removes
594         it from the socket set */
595 /**
596         @brief Close a socket, and remove it from the socket_manager's list.
597         @param mgr Pointer to the socket_manager.
598         @param sock_fd File descriptor for the socket to be closed.
599
600         We close the socket before determining whether it belongs to the socket_manager in question.
601 */
602 void socket_disconnect(socket_manager* mgr, int sock_fd) {
603         osrfLogInternal( OSRF_LOG_MARK, "Closing socket %d", sock_fd);
604         close( sock_fd );
605         socket_remove_node(mgr, sock_fd);
606 }
607
608
609 /**
610         @brief Determine whether a socket is valid.
611         @param sock_fd File descriptor for the socket.
612         @return 1 if the socket is valid, or 0 if it isn't.
613
614         The test is based on a call to select().  If the socket is valid but is not ready to be
615         written to, we wait until it is ready, then return 1.
616
617         If the select() fails, it may be because it was interrupted by a signal.  In that case
618         we try again.  Otherwise we assume that the socket is no longer valid.  This can happen
619         if, for example, the other end of a connection has closed the connection.
620
621         The select() can also fail if it is unable to allocate enough memory for its own internal
622         use.  If that happens, we may erroneously report a valid socket as invalid, but we
623         probably wouldn't be able to use it anyway if we're that close to exhausting memory.
624 */
625 int socket_connected(int sock_fd) {
626         fd_set read_set;
627         FD_ZERO( &read_set );
628         FD_SET( sock_fd, &read_set );
629         while( 1 ) {
630                 if( select( sock_fd + 1, &read_set, NULL, NULL, NULL) == -1 )
631                         return 0;
632                 else if( EINTR == errno )
633                         continue;
634                 else
635                         return 1;
636         }
637 }
638
639 /**
640         @brief Look for input on a given socket.  If you find some, react to it.
641         @param mgr Pointer to the socket_manager that presumably owns the socket.
642         @param timeout Timeout interval, in seconds (see notes).
643         @param sock_fd The file descriptor to look at.
644         @return 0 if successful, or -1 if a timeout or other error occurs.
645
646         If @a timeout is -1, wait indefinitely for input activity to appear.  If @a timeout is
647         zero, don't wait at all.  If @a timeout is positive, wait that number of seconds
648         before timing out.  If @a timeout has a negative value other than -1, the results are not
649         well defined, but we'll probably get an EINVAL error from select().
650
651         If we detect activity, branch on the type of socket:
652
653         - If it's a listener, accept a new connection, and add the new socket to the
654         socket_manager's list, without actually reading any data.
655         - Otherwise, read as much data as is available from the input socket, passing it a
656         buffer at a time to whatever callback function has been defined to the socket_manager.
657 */
658 int socket_wait( socket_manager* mgr, int timeout, int sock_fd ) {
659
660         int retval = 0;
661         fd_set read_set;
662         FD_ZERO( &read_set );
663         FD_SET( sock_fd, &read_set );
664
665         struct timeval tv;
666         tv.tv_sec = timeout;
667         tv.tv_usec = 0;
668         errno = 0;
669
670         if( timeout < 0 ) {
671
672                 // If timeout is -1, we block indefinitely
673                 if( (retval = select( sock_fd + 1, &read_set, NULL, NULL, NULL)) == -1 ) {
674                         osrfLogDebug( OSRF_LOG_MARK, "Call to select() interrupted: Sys Error: %s",
675                                         strerror(errno));
676                         return -1;
677                 }
678
679         } else if( timeout > 0 ) { /* timeout of 0 means don't block */
680
681                 if( (retval = select( sock_fd + 1, &read_set, NULL, NULL, &tv)) == -1 ) {
682                         osrfLogDebug( OSRF_LOG_MARK, "Call to select() interrupted: Sys Error: %s",
683                                         strerror(errno));
684                         return -1;
685                 }
686         }
687
688         osrfLogInternal( OSRF_LOG_MARK, "%d active sockets after select()", retval);
689
690         socket_node* node = socket_find_node(mgr, sock_fd);
691         if( node ) {
692                 if( node->endpoint == SERVER_SOCKET ) {
693                         _socket_handle_new_client( mgr, node );  // accept new connection
694                 } else {
695                         int status = _socket_handle_client_data( mgr, node );   // read data
696                         if( status == -1 ) {
697                                 socket_remove_node(mgr, sock_fd);
698                                 return -1;
699                         }
700                 }
701                 return 0;
702         }
703         else
704                 return -1;    // No such file descriptor for this socket_manager
705 }
706
707
708 int socket_wait_all(socket_manager* mgr, int timeout) {
709
710         if(mgr == NULL) {
711                 osrfLogWarning( OSRF_LOG_MARK,  "socket_wait_all(): null mgr" );
712                 return -1;
713         }
714
715         int retval = 0;
716         fd_set read_set;
717         FD_ZERO( &read_set );
718
719         socket_node* node = mgr->socket;
720         int max_fd = 0;
721         while(node) {
722                 osrfLogInternal( OSRF_LOG_MARK, "Adding socket fd %d to select set",node->sock_fd);
723                 FD_SET( node->sock_fd, &read_set );
724                 if(node->sock_fd > max_fd) max_fd = node->sock_fd;
725                 node = node->next;
726         }
727         max_fd += 1;
728
729         struct timeval tv;
730         tv.tv_sec = timeout;
731         tv.tv_usec = 0;
732         errno = 0;
733
734         if( timeout < 0 ) {  
735
736                 // If timeout is -1, there is no timeout passed to the call to select
737                 if( (retval = select( max_fd, &read_set, NULL, NULL, NULL)) == -1 ) {
738                         osrfLogWarning( OSRF_LOG_MARK, "select() call aborted: %s", strerror(errno));
739                         return -1;
740                 }
741
742         } else if( timeout != 0 ) { /* timeout of 0 means don't block */
743
744                 if( (retval = select( max_fd, &read_set, NULL, NULL, &tv)) == -1 ) {
745                         osrfLogWarning( OSRF_LOG_MARK, "select() call aborted: %s", strerror(errno));
746                         return -1;
747                 }
748         }
749
750         osrfLogDebug( OSRF_LOG_MARK, "%d active sockets after select()", retval);
751         return _socket_route_data(mgr, retval, &read_set);
752 }
753
754 /* iterates over the sockets in the set and handles active sockets.
755         new sockets connecting to server sockets cause the creation
756         of a new socket node.
757         Any new data read is is passed off to the data_received callback
758         as it arrives */
759 /* determines if we're receiving a new client or data
760         on an existing client */
761 static int _socket_route_data(
762         socket_manager* mgr, int num_active, fd_set* read_set) {
763
764         if(!(mgr && read_set)) return -1;
765
766         int last_failed_id = -1;
767
768
769         /* come back here if someone yanks a socket_node from beneath us */
770         while(1) {
771
772                 socket_node* node = mgr->socket;
773                 int handled = 0;
774                 int status = 0;
775                 
776                 while(node && (handled < num_active)) {
777         
778                         int sock_fd = node->sock_fd;
779                         
780                         if(last_failed_id != -1) {
781                                 /* in case it was not removed by our overlords */
782                                 osrfLogInternal( OSRF_LOG_MARK, "Attempting to remove last_failed_id of %d", last_failed_id);
783                                 socket_remove_node( mgr, last_failed_id );
784                                 last_failed_id = -1;
785                                 status = -1;
786                                 break;
787                         }
788         
789                         /* does this socket have data? */
790                         if( FD_ISSET( sock_fd, read_set ) ) {
791         
792                                 osrfLogInternal( OSRF_LOG_MARK, "Socket %d active", sock_fd);
793                                 handled++;
794                                 FD_CLR(sock_fd, read_set);
795         
796                                 if(node->endpoint == SERVER_SOCKET) 
797                                         _socket_handle_new_client(mgr, node);
798         
799                                 else
800                                         status = _socket_handle_client_data(mgr, node);
801         
802                                 /* someone may have yanked a socket_node out from under 
803                                         us...start over with the first socket */
804                                 if(status == -1)  {
805                                         last_failed_id = sock_fd;
806                                         osrfLogInternal( OSRF_LOG_MARK, "Backtracking back to start of loop because "
807                                                         "of -1 return code from _socket_handle_client_data()");
808                                 }
809                         }
810
811                         if(status == -1) break;
812                         node = node->next;
813
814                 } // is_set
815
816                 if(status == 0) break;
817                 if(status == -1) status = 0;
818         } 
819
820         return 0;
821 }
822
823
824 /**
825         @brief Accept a new socket from a listener, and add it to the socket_manager's list.
826         @param mgr Pointer to the socket_manager that will own the new socket.
827         @param node Pointer to the socket_node for the listener socket.
828         @return 0 if successful, or -1 if not.
829
830         Call: accept().  Creates a CLIENT_SOCKET (even though the socket resides on the server).
831 */
832 static int _socket_handle_new_client(socket_manager* mgr, socket_node* node) {
833         if(mgr == NULL || node == NULL) return -1;
834
835         errno = 0;
836         int new_sock_fd;
837         new_sock_fd = accept(node->sock_fd, NULL, NULL);
838         if(new_sock_fd < 0) {
839                 osrfLogWarning( OSRF_LOG_MARK, "_socket_handle_new_client(): accept() failed: %s",
840                         strerror( errno ) );
841                 return -1;
842         }
843
844         if(node->addr_type == INET) {
845                 _socket_add_node(mgr, CLIENT_SOCKET, INET, new_sock_fd, node->sock_fd);
846                 osrfLogDebug( OSRF_LOG_MARK, "Adding new INET client for %d", node->sock_fd);
847
848         } else if(node->addr_type == UNIX) {
849                 _socket_add_node(mgr, CLIENT_SOCKET, UNIX, new_sock_fd, node->sock_fd);
850                 osrfLogDebug( OSRF_LOG_MARK, "Adding new UNIX client for %d", node->sock_fd);
851         }
852
853         return 0;
854 }
855
856
857 /**
858         @brief Receive data on a streaming socket.
859         @param mgr Pointer to the socket_manager that owns the socket_node.
860         @param node Pointer to the socket_node that owns the socket.
861         @return 0 if successful, or -1 upon failure.
862
863         Receive one or more buffers until no more bytes are available for receipt.  Add a
864         terminal nul to each buffer and pass it to a callback function previously defined by the
865         application to the socket_manager.
866
867         If the sender closes the connection, call another callback function, if one has been
868         defined.
869
870         Even when the function returns successfully, the received message may not be complete --
871         there may be more data that hasn't arrived yet.  It is the responsibility of the
872         calling code to recognize message boundaries.
873
874         Called only for a CLIENT_SOCKET.
875 */
876 static int _socket_handle_client_data(socket_manager* mgr, socket_node* node) {
877         if(mgr == NULL || node == NULL) return -1;
878
879         char buf[RBUFSIZE];
880         int read_bytes;
881         int sock_fd = node->sock_fd;
882
883         set_fl(sock_fd, O_NONBLOCK);
884
885         osrfLogInternal( OSRF_LOG_MARK, "%ld : Received data at %f\n",
886                         (long) getpid(), get_timestamp_millis());
887
888         while( (read_bytes = recv(sock_fd, buf, RBUFSIZE-1, 0) ) > 0 ) {
889                 buf[read_bytes] = '\0';
890                 osrfLogInternal( OSRF_LOG_MARK, "Socket %d Read %d bytes and data: %s",
891                                 sock_fd, read_bytes, buf);
892                 if(mgr->data_received)
893                         mgr->data_received(mgr->blob, mgr, sock_fd, buf, node->parent_id);
894         }
895         int local_errno = errno; /* capture errno as set by recv() */
896
897         if(socket_find_node(mgr, sock_fd)) {  /* someone may have closed this socket */
898                 clr_fl(sock_fd, O_NONBLOCK);
899                 if(read_bytes < 0) {
900                         // EAGAIN would have meant that no more data was available
901                         if(local_errno != EAGAIN)   // but if that's not the case...
902                                 osrfLogWarning( OSRF_LOG_MARK, " * Error reading socket with error %s",
903                                         strerror(local_errno) );
904                 }
905
906         } else { return -1; } /* inform the caller that this node has been tampered with */
907
908         if(read_bytes == 0) {  /* socket closed by client */
909                 if(mgr->on_socket_closed) {
910                         mgr->on_socket_closed(mgr->blob, sock_fd);
911                 }
912                 return -1;
913         }
914
915         return 0;
916
917 }
918
919
920 /**
921         @brief Destroy a socket_manager, and close all of its sockets.
922         @param mgr Pointer to the socket_manager to be destroyed.
923 */
924 void socket_manager_free(socket_manager* mgr) {
925         if(mgr == NULL) return;
926         socket_node* tmp;
927         while(mgr->socket) {
928                 tmp = mgr->socket->next;
929                 socket_disconnect(mgr, mgr->socket->sock_fd);
930                 mgr->socket = tmp;
931         }
932         free(mgr);
933
934 }
935