some code cleanup, some unfinished projects, etc.
[OpenSRF.git] / 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 int debug_handler(char* msg, ...) {
43         va_list args;
44         va_start(args, msg);
45         vfprintf(stderr, msg, args);
46         va_end(args);
47         fprintf( stderr, "\n" );
48         return -1;
49 }
50
51 int warning_handler(char* msg, ...) {
52         va_list args;
53         va_start(args, msg);
54         vfprintf(stderr, msg, args);
55         va_end(args);
56         fprintf( stderr, "\n" );
57         return -1;
58 }
59 */
60
61
62 socket_node* _socket_add_node(socket_manager* mgr, 
63                 int endpoint, int addr_type, int sock_fd, int parent_id ) {
64
65         if(mgr == NULL) return NULL;
66         debug_handler("Adding socket node with fd %d", sock_fd);
67         socket_node* new_node = safe_malloc(sizeof(socket_node));
68
69         new_node->endpoint      = endpoint;
70         new_node->addr_type     = addr_type;
71         new_node->sock_fd               = sock_fd;
72         new_node->next                  = NULL;
73         new_node->parent_id = 0;
74         if(parent_id > 0)
75                 new_node->parent_id = parent_id;
76
77         new_node->next                  = mgr->socket;
78         mgr->socket                             = new_node;
79         return new_node;
80 }
81
82 /* creates a new server socket node and adds it to the socket set.
83         returns new socket fd on success.  -1 on failure.
84         socket_type is one of INET or UNIX  */
85 int socket_open_tcp_server(socket_manager* mgr, int port) {
86
87         if( mgr == NULL ) return warning_handler("socket_open_tcp_server(): NULL mgr"); 
88
89         int sock_fd;
90         struct sockaddr_in server_addr;
91
92         sock_fd = socket(AF_INET, SOCK_STREAM, 0);
93
94         if(sock_fd < 0) 
95                 return warning_handler("tcp_server_connect(): Unable to create socket");
96
97         server_addr.sin_family = AF_INET;
98         server_addr.sin_addr.s_addr = htonl(INADDR_ANY);
99         server_addr.sin_port = htons(port);
100
101         if(bind( sock_fd, (struct sockaddr*) &server_addr, sizeof(server_addr)) < 0)
102                 return warning_handler("tcp_server_connect(): cannot bind to port %d", port );
103
104         if(listen(sock_fd, 20) == -1) 
105                 return warning_handler("tcp_server_connect(): listen() returned error");
106
107         _socket_add_node(mgr, SERVER_SOCKET, INET, sock_fd, 0);
108         return sock_fd;
109 }
110
111 int socket_open_unix_server(socket_manager* mgr, char* path) {
112         if(mgr == NULL || path == NULL) return -1;
113
114         debug_handler("opening unix socket at %s", path);
115         int sock_fd;
116         struct sockaddr_un server_addr;
117
118         sock_fd = socket(AF_UNIX, SOCK_STREAM, 0);
119         if(sock_fd < 0)
120                 return warning_handler("socket_open_unix_server(): socket() failed");
121
122         server_addr.sun_family = AF_UNIX;
123         strcpy(server_addr.sun_path, path);
124
125         if( bind(sock_fd, (struct sockaddr*) &server_addr, 
126                                 sizeof(struct sockaddr_un)) < 0) {
127                 return warning_handler(
128                         "socket_open_unix_server(): cannot bind to unix port %s", path );
129         }
130
131         if(listen(sock_fd, 20) == -1) 
132                 return warning_handler("socket_open_unix_server(): listen() returned error");
133
134         debug_handler("unix socket successfully opened");
135         _socket_add_node(mgr, SERVER_SOCKET, UNIX, sock_fd, 0);
136         return sock_fd;
137 }
138
139
140
141 /* creates a client socket and adds it to the socket set.
142         returns 0 on success.  -1 on failure.
143         socket_type is one of INET or UNIX  */
144 int socket_open_client(socket_manager* mgr, 
145                 int socket_type, int port, char* sock_path, char* dest_addr) {
146         return 0;
147 }
148
149 /* returns the socket_node with the given sock_fd */
150 socket_node* socket_find_node(socket_manager* mgr, int sock_fd) {
151         if(mgr == NULL) return NULL;
152         socket_node* node = mgr->socket;
153         while(node) {
154                 if(node->sock_fd == sock_fd)
155                         return node;
156                 node = node->next;
157         }
158         return NULL;
159 }
160
161 /* removes the node with the given sock_fd from the list and frees it */
162 void socket_remove_node(socket_manager* mgr, int sock_fd) {
163
164         if(mgr == NULL) return;
165
166         debug_handler("removing socket %d", sock_fd);
167
168         socket_node* head = mgr->socket;
169         socket_node* tail = head;
170         if(head == NULL) return;
171
172         /* if removing the first node in the list */
173         if(head->sock_fd == sock_fd) {
174                 mgr->socket = head->next;
175                 free(head);
176                 debug_handler("removing first socket in list");
177                 return;
178         }
179
180         head = head->next;
181
182         /* if removing any other node */
183         while(head) {
184                 if(head->sock_fd == sock_fd) {
185                         tail->next = head->next;
186                         free(head);
187                         return;
188                 }
189                 tail = head;
190                 head = head->next;
191         }
192 }
193
194
195
196 void _socket_print_list(socket_manager* mgr) {
197         if(mgr == NULL) return;
198         socket_node* node = mgr->socket;
199         debug_handler("socket_node list: [");
200         while(node) {
201                 debug_handler("sock_fd: %d | parent_id: %d", 
202                                 node->sock_fd, node->parent_id);
203                 node = node->next;
204         }
205         debug_handler("]");
206 }
207
208 /* sends the given data to the given socket */
209 int socket_send(int sock_fd, const char* data) {
210         debug_handler( "socket_bundle sending to %d data %s",
211                 sock_fd, data);
212
213         signal(SIGPIPE, SIG_IGN); /* in case a unix socket was closed */
214         if( send( sock_fd, data, strlen(data), 0 ) < 0 ) {
215                 return warning_handler( "tcp_server_send(): Error sending data" );
216         }
217
218         return 0;
219 }
220
221 /* disconnects the node with the given sock_fd and removes
222         it from the socket set */
223 void socket_disconnect(socket_manager* mgr, int sock_fd) {
224
225         debug_handler("Closing socket %d", sock_fd);
226
227         if( shutdown( sock_fd, SHUT_RDWR ) )
228                 warning_handler( "socket_disconnect(): Error shuting down socket, removing anyway" );
229
230         if( close( sock_fd ) == -1 ) 
231                 warning_handler( "socket_disconnect(): Error closing socket, removing anyway" );
232
233         if(mgr != NULL) 
234                 socket_remove_node(mgr, sock_fd);
235         
236 }
237
238
239 /* we assume that if select() fails, the socket is no longer valid */
240 int socket_connected(int sock_fd) {
241         fd_set read_set;
242         FD_ZERO( &read_set );
243         FD_SET( sock_fd, &read_set );
244         if( select( sock_fd + 1, &read_set, NULL, NULL, NULL) == -1 ) 
245                 return 0;
246         return 1;
247
248 }
249
250 int socket_wait(socket_manager* mgr, int timeout, int sock_fd) {
251         return 0;
252 }
253
254
255 int socket_wait_all(socket_manager* mgr, int timeout) {
256
257         if(mgr == NULL) return warning_handler( "tcp_wait(): null mgr" );
258
259         int retval = 0;
260         fd_set read_set;
261         FD_ZERO( &read_set );
262
263         socket_node* node = mgr->socket;
264         int max_fd = 0;
265         while(node) {
266                 //debug_handler("Adding socket %d to select set",node->sock_fd);
267                 FD_SET( node->sock_fd, &read_set );
268                 if(node->sock_fd > max_fd) max_fd = node->sock_fd;
269                 node = node->next;
270         }
271         max_fd += 1;
272
273         struct timeval tv;
274         tv.tv_sec = timeout;
275         tv.tv_usec = 0;
276
277         if( timeout == -1 ) {  
278
279                 // If timeout is -1, there is no timeout passed to the call to select
280                 if( (retval = select( max_fd, &read_set, NULL, NULL, NULL)) == -1 ) {
281                         return warning_handler("Call to select interrupted");
282                 }
283
284         } else if( timeout != 0 ) { /* timeout of 0 means don't block */
285
286                 if( (retval = select( max_fd, &read_set, NULL, NULL, &tv)) == -1 ) {
287                         return warning_handler( "Call to select interrupted" );
288                 }
289         }
290
291         debug_handler("%d active sockets after select()", retval);
292         return _socket_route_data(mgr, retval, &read_set);
293 }
294
295 /* determines if we'er receiving a new client or data
296         on an existing client */
297 int _socket_route_data(
298         socket_manager* mgr, int num_active, fd_set* read_set) {
299
300         if(mgr == NULL) return -1;
301
302         int last_failed_id = -1;
303
304
305         /* come back here if someone yanks a socket_node from beneath us */
306         while(1) {
307
308                 socket_node* node = mgr->socket;
309                 int handled = 0;
310                 int status = 0;
311                 
312                 while(node && (handled < num_active)) {
313         
314                         int sock_fd = node->sock_fd;
315                         
316                         if(last_failed_id != -1) {
317                                 /* in case it was not removed by our overlords */
318                                 debug_handler("Attempting to remove last_failed_id of %d", last_failed_id);
319                                 socket_remove_node( mgr, last_failed_id );
320                                 last_failed_id = -1;
321                                 status = -1;
322                                 break;
323                         }
324
325         
326                         /* does this socket have data? */
327                         if( FD_ISSET( sock_fd, read_set ) ) {
328
329         
330                                 debug_handler("Socket %d active", sock_fd);
331                                 handled++;
332                                 FD_CLR(sock_fd, read_set);
333         
334                                 if(node->endpoint == SERVER_SOCKET) 
335                                         _socket_handle_new_client(mgr, node);
336         
337                                 if(node->endpoint == CLIENT_SOCKET ) 
338                                         status = _socket_handle_client_data(mgr, node);
339         
340                                 /* someone may have yanked a socket_node out from under 
341                                         us...start over with the first socket */
342                                 if(status == -1)  {
343                                         last_failed_id = sock_fd;
344                                         debug_handler("Backtracking back to start of loop because "
345                                                         "of -1 return code from _socket_handle_client_data()");
346                                 }
347                         }
348
349                         if(status == -1) break;
350                         node = node->next;
351
352                 } // is_set
353
354                 if(status == 0) break;
355                 if(status == -1) status = 0;
356         } 
357
358         return 0;
359 }
360
361
362 int _socket_handle_new_client(socket_manager* mgr, socket_node* node) {
363         if(mgr == NULL || node == NULL) return -1;
364
365         //struct sockaddr_in client_addr_in; 
366         //struct sockaddr_un client_addr_un; 
367         //int client_len, new_sock_fd; 
368
369         int new_sock_fd;
370         new_sock_fd = accept(node->sock_fd, NULL, NULL);
371         if(new_sock_fd < 0)
372                 return warning_handler("_socket_route_data(): accept() failed");
373
374         if(node->addr_type == INET) {
375                 _socket_add_node(mgr, CLIENT_SOCKET, INET, new_sock_fd, node->sock_fd);
376                 debug_handler("Adding new INET client for %d", node->sock_fd);
377
378         } else if(node->addr_type == UNIX) {
379                 _socket_add_node(mgr, CLIENT_SOCKET, UNIX, new_sock_fd, node->sock_fd);
380                 debug_handler("Adding new UNIX client for %d", node->sock_fd);
381         }
382
383         return 0;
384 }
385
386
387 int _socket_handle_client_data(socket_manager* mgr, socket_node* node) {
388         if(mgr == NULL || node == NULL) return -1;
389
390         char buf[BUFSIZE];
391         int read_bytes;
392         int sock_fd = node->sock_fd;
393
394         memset(buf, 0, BUFSIZE);
395         set_fl(sock_fd, O_NONBLOCK);
396         debug_handler("Gathering client data for %d", node->sock_fd);
397
398         debug_handler("Socket buf before read %s", buf);
399         while( (read_bytes = recv(sock_fd, buf, BUFSIZE-1, 0) ) > 0 ) {
400                 debug_handler("Socket %d Read %d bytes and data: %s", sock_fd, read_bytes, buf);
401                 if(mgr->data_received)
402                         mgr->data_received(mgr->blob, mgr, sock_fd, buf, node->parent_id);
403
404                 memset(buf, 0, BUFSIZE);
405         }
406
407         if(socket_find_node(mgr, sock_fd)) {  /* someone may have closed this socket */
408                 clr_fl(sock_fd, O_NONBLOCK); 
409                 if(read_bytes < 0) { 
410                         if( errno != EAGAIN ) 
411                                 warning_handler( " * Error reading socket with errno %d", errno );
412                 }
413         }
414
415         if(read_bytes == 0) {  /* socket closed by client */
416                 if(mgr->on_socket_closed) {
417                         mgr->on_socket_closed(mgr->blob, sock_fd);
418                         return -1;
419                 }
420         }
421
422         return 0;
423
424 }
425
426
427 void socket_manager_free(socket_manager* mgr) {
428         if(mgr == NULL) return;
429         socket_node* tmp;
430         while(mgr->socket) {
431                 tmp = mgr->socket->next;
432                 socket_disconnect(mgr, mgr->socket->sock_fd);
433                 mgr->socket = tmp;
434         }
435         free(mgr);
436
437 }