changed router binary to opensrf_router to prevent opensrf_all from destroying
[OpenSRF.git] / src / router / router.c
1 #include "router.h"
2 #include <sys/types.h>
3 #include <signal.h>
4
5 #define ROUTER_MAX_MSGS_PER_PACKET 256
6 char* router_resource;
7 transport_router_registrar* routt;
8 void _build_trusted_sites( transport_router_registrar* router );
9
10
11 void sig_hup_handler( int a ) { 
12         router_registrar_free( routt ); 
13         osrfConfigCleanup();
14         log_free();
15         free( router_resource );
16         exit(0); 
17 }
18
19
20 int main( int argc, char* argv[] ) {
21
22         if( argc < 2 ) {
23                 fatal_handler( "Usage: %s <path_to_config_file>", argv[0] );
24                 exit(0);
25         }
26
27         osrfConfig* cfg = osrfConfigInit( argv[1], "router" );
28         osrfConfigSetDefaultConfig(cfg);
29
30         init_proc_title( argc, argv );
31         set_proc_title( "OpenSRF Router" );
32
33         /* load the config options */
34         char* server                    = osrfConfigGetValue(NULL, "/transport/server");
35         char* port                              = osrfConfigGetValue(NULL, "/transport/port");
36         char* unixpath                  = osrfConfigGetValue(NULL, "/transport/unixpath");
37         char* username                  = osrfConfigGetValue(NULL, "/transport/username");
38         char* password                  = osrfConfigGetValue(NULL, "/transport/password");
39         router_resource         = osrfConfigGetValue(NULL, "/transport/resource");
40         char* con_timeout               = osrfConfigGetValue(NULL, "/transport/connect_timeout" );
41         char* max_retries               = osrfConfigGetValue(NULL, "/transport/max_reconnect_attempts" );
42         char* component         = osrfConfigGetValue(NULL, "/component" );
43
44         fprintf(stderr, "Router connecting with uname %s, server %s, port %s, unixpath %s",
45                 username, server, port, unixpath );
46
47         /* set up the logger */
48         char* level = osrfConfigGetValue(NULL, "/loglevel");
49         char* log_file = osrfConfigGetValue(NULL, "/logfile");
50
51         int llevel = atoi(level);
52         fprintf(stderr, "Level %d; file %s\n", llevel, log_file );
53
54         if(!log_init( llevel, log_file )) 
55                 fprintf(stderr, "Unable to init logging, going to stderr...\n" );
56
57
58         free(level);
59         free(log_file);
60
61         fprintf(stderr, "Router connecting as \nserver: %s \nport: %s \nunixpath: %s\nuser:%s \nresource:%s\n", 
62                         server, port, unixpath, username, router_resource );
63
64         int iport = 0;
65         if(port)        iport = atoi( port );
66
67         int con_itimeout        = atoi( con_timeout );
68         int max_retries_        = atoi(max_retries);
69         int icomponent = 0;
70         if(component) 
71                 icomponent = atoi(component);
72
73         /* build the router_registrar */
74         transport_router_registrar* router_registrar = 
75                 router_registrar_init( server, iport, unixpath, username, password, router_resource, 0, con_itimeout, icomponent ); 
76
77         routt = router_registrar;
78
79         free(server);
80         free(port);
81         free(unixpath);
82         free(username);
83         free(password);
84         free(con_timeout);
85         free(max_retries);
86
87         daemonize();
88
89         signal(SIGHUP,sig_hup_handler);
90         signal(SIGINT,sig_hup_handler);
91         signal(SIGTERM,sig_hup_handler);
92
93
94         int counter = 0;
95         /* wait for incoming... */
96         while( ++counter <= max_retries_ ) {
97
98                 /* connect to jabber */
99                 if( router_registrar_connect( router_registrar ) )  {
100                         info_handler( "Connected..." );
101                         fprintf(stderr, "- Connected -\n");
102                         counter = 0;
103                         listen_loop( router_registrar );
104                 } else  
105                         fatal_handler( "Could not connect to Jabber Server" );
106
107                 /* this should never happen */
108                 warning_handler( "Jabber server probably went away, attempting reconnect" );
109
110                 sleep(5);
111         }
112
113
114         router_registrar_free( router_registrar );
115         return 1;
116
117 }
118
119 transport_router_registrar* router_registrar_init( char* server, 
120                 int port, char* unixpath, char* username, char* password, 
121                 char* resource, int client_timeout, int con_timeout, int component ) {
122
123         if( server == NULL ) { return NULL; }
124         
125         /* allocate a new router_registrar object */
126         size_t size = sizeof( transport_router_registrar );
127         transport_router_registrar* router_registrar = (transport_router_registrar*) safe_malloc( size );
128
129         router_registrar->client_timeout        = client_timeout;
130         router_registrar->jabber = jabber_connect_init( server, port, unixpath, username, password, resource, con_timeout, component );
131         _build_trusted_sites( router_registrar );
132
133         return router_registrar;
134 }
135
136 void _build_trusted_sites( transport_router_registrar* router ) {
137
138         router->trusted_servers = osrfNewStringArray(4);
139         router->trusted_clients = osrfNewStringArray(4);
140         osrfConfigGetValueList(NULL, router->trusted_servers, "/trusted_domains/server" );
141         osrfConfigGetValueList(NULL, router->trusted_clients, "/trusted_domains/client" );
142
143         if(router->trusted_servers->size < 1 ||
144                 router->trusted_clients->size < 1 )
145                 fatal_handler( "You must specify at least one trusted server and client in the config file");
146 }
147
148
149 jabber_connect* jabber_connect_init(    
150         char* server, int port, char* unixpath, char* username, 
151                 char* password, char* resource, int connect_timeout, int component ) {
152
153         size_t len = sizeof(jabber_connect);
154         jabber_connect* jabber = (jabber_connect*) safe_malloc( len );
155
156         jabber->port                            = port;
157         jabber->connect_timeout = connect_timeout;
158         jabber->unixpath                        = strdup(unixpath);
159
160         jabber->server                          = strdup(server);
161         jabber->username                        = strdup(username);
162         jabber->password                        = strdup(password);
163         jabber->resource                        = strdup(resource);
164
165         if( jabber->server == NULL || jabber->username == NULL ||
166                         jabber->password == NULL || jabber->resource == NULL ) {
167                 fatal_handler( "jabber_init(): Out of Memory" );
168                 return NULL;
169         }
170
171         /* build the transport client */
172         jabber->t_client = client_init( jabber->server, jabber->port, unixpath, component );
173
174         return jabber;
175 }
176
177 /* connect the router_registrar to jabber */
178 int router_registrar_connect( transport_router_registrar* router ) {
179         return j_connect( router->jabber );
180 }
181
182 /* connect a jabber_connect object jabber */
183 int j_connect( jabber_connect* jabber ) {
184         if( jabber == NULL ) { return 0; }
185         return client_connect( jabber->t_client, 
186                         jabber->username, jabber->password, jabber->resource, 
187                         jabber->connect_timeout, AUTH_DIGEST );
188 }
189
190 int fill_fd_set( transport_router_registrar* router, fd_set* set ) {
191         
192         int max_fd;
193         FD_ZERO(set);
194
195         int router_fd = router->jabber->t_client->session->sock_id;
196         max_fd = router_fd;
197         FD_SET( router_fd, set );
198
199         server_class_node* cur_node = router->server_class_list;
200         while( cur_node != NULL ) {
201                 int cur_class_fd = cur_node->jabber->t_client->session->sock_id;
202                 if( cur_class_fd > max_fd ) 
203                         max_fd = cur_class_fd;
204                 FD_SET( cur_class_fd, set );
205                 cur_node = cur_node->next;
206         }
207
208         FD_CLR( 0, set );
209         return max_fd;
210 }
211
212
213 void listen_loop( transport_router_registrar* router ) {
214
215         if( router == NULL )
216                 return;
217
218         int select_ret;
219         int router_fd = router->jabber->t_client->session->sock_id;
220         transport_message* cur_msg;
221
222         while(1) {
223
224                 fd_set listen_set;
225                 int max_fd = fill_fd_set( router, &listen_set );
226
227                 if( max_fd < 1 ) 
228                         fatal_handler( "fill_fd_set return bogus max_fd: %d", max_fd );
229
230                 int num_handled = 0;
231                 info_handler( "Router going into select() wait..." );
232
233                 if( (select_ret=select(max_fd+ 1, &listen_set, NULL, NULL, NULL)) < 0 ) {
234
235                         warning_handler( "Select returned error %d", select_ret );
236                         warning_handler( "Select Error %d on fd %d", errno );
237                         perror( "Select Error" );
238                         warning_handler( "Errors: EBADF %d, EINTR %d, EINVAL %d, ENOMEM %d",
239                                         EBADF, EINTR, EINVAL, ENOMEM );
240                         continue;
241
242                 } else {
243
244                         info_handler( "Select returned %d", select_ret );
245                         
246                         if( FD_ISSET( router_fd, &listen_set ) ) {
247                                 cur_msg = client_recv( router->jabber->t_client, 1 );
248
249                                 /* We only process a message if we have some trusted servers and the current
250                                         message came from one of those servers */
251                                 if(cur_msg) {
252
253                                         if( router->trusted_servers->size > 0 ) {
254
255                                                 int i = 0;
256                                                 int found = 0;
257         
258                                                 if( cur_msg->sender ) {
259
260                                                         int len = strlen(cur_msg->sender) + 1; /* there's no way it could be that big, but... */
261                                                         char server_buf[len];
262                                                         memset(server_buf,0,len);
263                                                         jid_get_domain( cur_msg->sender, server_buf );
264                                                         info_handler("Received top level message from %s", server_buf );
265                 
266                                                         while(1) {
267                                                                 char* domain = osrfStringArrayGetString(router->trusted_servers, i);
268                                                                 if(domain) {
269                                                                         if(!strcmp(domain, server_buf)) {
270                                                                                 found = 1;
271                                                                                 break;
272                                                                         }
273                                                                 }
274                                                                 i++;
275                                                         }
276                                                 }
277
278                                                 if(found)
279                                                         router_registrar_handle_msg( router, cur_msg );
280                                                 else
281                                                         warning_handler( "Received top level message from unpriveleged sender %s", cur_msg->sender );
282                                         }
283         
284                                         message_free( cur_msg );
285                                 }
286
287                                 if( ++num_handled == select_ret ) 
288                                         continue;
289                         }
290
291                         /* cycle through the children and find any whose fd's are ready for reading */
292                         server_class_node* cur_node = router->server_class_list;
293                         while( cur_node != NULL ) {
294                                 debug_handler( "Checking File Descriptor" );
295                                 int cur_fd = cur_node->jabber->t_client->session->sock_id;
296
297                                 debug_handler("Router checking file descriptor %d", cur_fd);
298
299                                 if( FD_ISSET(cur_fd, &listen_set) ) {
300                                         ++num_handled;
301                                         FD_CLR(cur_fd,&listen_set);
302
303                                         debug_handler("Router has data on file descriptor %d", cur_fd);
304                                         cur_msg = client_recv( cur_node->jabber->t_client, 1 );
305
306                                         if(cur_msg) {
307
308                                                 info_handler( "%s received from %s", cur_node->server_class, cur_msg->sender );
309                                                 int handle_ret = server_class_handle_msg( router, cur_node, cur_msg );
310         
311                                                 if( handle_ret == -1 ) {
312                                                         warning_handler( "server_class_handle_msg() returned -1" );
313                                                         cur_node = router->server_class_list; /*start over*/
314                                                         continue;
315         
316                                                 } else if( handle_ret == 0 ) {
317                                                         /* delete and continue */
318                                                         warning_handler( "server_class_handle_msg() returned 0" );
319                                                         remove_server_class( router, cur_node );        
320                                                         debug_handler( "Removed Server Class" );
321                                                         cur_node = router->server_class_list; /*start over*/
322                                                         continue;
323                                                 } 
324         
325                                                 info_handler( "%s handled message successfully", cur_node->server_class );
326                                                 /* dont free message here */
327                                                 if( num_handled == select_ret ) 
328                                                         break;
329                                         }
330
331                                 }
332
333                                 if( num_handled == select_ret ) 
334                                         break;
335
336                                 cur_node = cur_node->next;
337
338                         } /* cycling through the server_class list */
339
340                 } /* no select errors */
341         } 
342 }
343
344
345 /* determine where to route top level messages */
346 int router_registrar_handle_msg( transport_router_registrar* router_registrar, transport_message* msg ) {
347
348         info_handler( "Received class: %s : command %s ", msg->router_class, msg->router_command );
349
350         if( router_registrar == NULL || msg == NULL ) { return 0; }
351
352         if( msg->router_command == NULL || !strcmp(msg->router_command,"") ) {
353                 return router_registrar_handle_app_request( router_registrar, msg );
354         }
355
356         // user issued a ruoter query
357         /* a query command has router_command="query" and the actual query type
358                 is the content of the message */
359         if( !strcmp(msg->router_command,"query")) {
360                 info_handler( "Router received query command" );
361
362                 // user issues a servers query
363                 if( !strcmp(msg->body, "servers")) {
364
365                         info_handler( "Router received servers query command" );
366                         router_return_server_info( router_registrar, msg );
367                         return 1;
368                 }
369         }
370
371
372         info_handler("Looking for server_class_node %s...",msg->router_class);
373         server_class_node* active_class_node = find_server_class( router_registrar, msg->router_class );
374
375         if( active_class_node == NULL ) { 
376                 info_handler("Could not find server_class_node %s, creating one.",msg->router_class);
377
378                 /* there is no server_class for msg->router_class so we build it here */
379                 if( strcmp( msg->router_command, "register") == 0 ) {
380
381                         info_handler("Adding server_class_node for %s",msg->router_class);
382                         active_class_node = 
383                                 init_server_class( router_registrar, msg->sender, msg->router_class ); 
384
385                         if( active_class_node == NULL ) {
386                                 fatal_handler( "router_listen(): active_class_node == NULL for %s", msg->sender );
387                                 return 0;
388                         }
389
390                         if (router_registrar->server_class_list != NULL) {
391                                 active_class_node->next = router_registrar->server_class_list;
392                                 router_registrar->server_class_list->prev = active_class_node;
393                         }
394                         router_registrar->server_class_list = active_class_node;
395
396                         //spawn_server_class( (void*) active_class_node );
397
398                 } else {
399                         warning_handler( "router_register_handler_msg(): Bad Command [%s] for class [%s]",
400                                 msg->router_command, msg->router_class );
401                 }
402
403         } else if( strcmp( msg->router_command, "register") == 0 ) {
404                 /* there is a server_class for msg->router_class so we 
405                         need to either add a new server_node or update the existing one */
406
407                 
408                 server_node* s_node = find_server_node( active_class_node, msg->sender );
409
410                 if( s_node != NULL ) {
411                         s_node->available = 1;
412                         s_node->upd_time = time(NULL);
413                         info_handler( "Found matching registered server: %s. Updating.",
414                                         s_node->remote_id );
415                 } else {
416                         s_node = init_server_node( msg->sender );
417
418                         info_handler( "Adding server_node for: %s.", s_node->remote_id );
419
420                         if (s_node == NULL ) {
421                                 warning_handler( " Could not create new xerver_node for %s.",
422                                         msg->sender );
423                                 return 0;
424                         }
425
426                         s_node->next = active_class_node->current_server_node->next;
427                         s_node->prev = active_class_node->current_server_node;
428
429                         active_class_node->current_server_node->next->prev = s_node;
430                         active_class_node->current_server_node->next = s_node;
431                 }
432
433
434         } else if( strcmp( msg->router_command, "unregister") == 0 ) {
435
436                 if( ! unregister_server_node( active_class_node, msg->sender ) ) {
437                         remove_server_class( router_registrar, active_class_node );
438                         debug_handler( "Removed server class after final unregister");
439                 }
440
441         } else {
442                 warning_handler( "router_register_handler_msg(): Bad Command [%s] for class [%s]",
443                         msg->router_command, msg->router_class );
444         }
445
446         return 1;
447 }
448
449
450 /* removes a server class node from the top level router_registrar */
451 int unregister_server_node( server_class_node* active_class_node, char* remote_id ) {
452
453         server_node* d_node = find_server_node( active_class_node, remote_id );
454
455         if ( d_node != NULL ) {
456
457                 info_handler( "Removing server_node for: %s.", d_node->remote_id );
458
459                 if ( d_node->next == NULL ) {
460                         warning_handler( "NEXT is NULL in ring [%s] -- "
461                                 "THIS SHOULD NEVER HAPPEN",
462                                 d_node->remote_id );
463
464                 }
465                 
466                 if ( d_node->prev == NULL ) {
467                         warning_handler( "PREV is NULL in a ring [%s] -- "
468                                 "THIS SHOULD NEVER HAPPEN",
469                                 d_node->remote_id );
470
471                 }
472
473                 if ( d_node->next == d_node && d_node->prev == d_node) {
474                         info_handler( "Last node, setting ring to NULL: %s.",
475                                 d_node->remote_id );
476
477                         active_class_node->current_server_node = NULL;
478
479                         server_node_free( d_node );
480                         return 0;
481
482                 } else {
483                         info_handler( "Nodes remain, splicing: %s, %s",
484                                 d_node->prev->remote_id,
485                                 d_node->next->remote_id);
486
487                 info_handler( "d_node => %x, next => %x, prev => %x",
488                                         d_node, d_node->next, d_node->prev );
489
490
491                         d_node->prev->next = d_node->next;
492                         d_node->next->prev = d_node->prev;
493
494                         info_handler( "prev => %x, prev->next => %x, prev->prev => %x",
495                                 d_node->prev, d_node->prev->next, d_node->prev->prev );
496
497                         info_handler( "next => %x, next->next => %x, next->prev => %x",
498                                 d_node->next, d_node->next->next, d_node->next->prev );
499                                 
500                         if (active_class_node->current_server_node == d_node)
501                                 active_class_node->current_server_node = d_node->next;
502
503
504                         server_node_free( d_node );
505                 }
506         } 
507
508         return 1;
509 }
510
511 server_node * find_server_node ( server_class_node * class, const char * remote_id ) {
512
513         if ( class == NULL ) {
514                 warning_handler(" find_server_node(): bad arg!");
515                 return NULL;
516         }
517
518         server_node * start_node = class->current_server_node;
519         server_node * node = class->current_server_node;
520
521         do {
522                 if (node == NULL)
523                         return NULL;
524
525                 if ( strcmp(node->remote_id, remote_id) == 0 )
526                         return node;
527
528                 node = node->next;
529
530         } while ( node != start_node );
531
532         return NULL;
533 }
534
535 /* if we return -1, then we just deleted the server_class you were looking for
536         if we return 0, then some other error has occured
537         we return 1 otherwise */
538 int remove_server_class( transport_router_registrar* router, server_class_node* class ) {
539         if( class == NULL )
540                 return 0;
541
542         transport_message * msg = NULL;
543         while ( (msg = client_recv(class->jabber->t_client, 0)) != NULL ) {
544                 debug_handler( "Looping on messages to dead class" );
545                 if( server_class_handle_msg(router, class, msg) < 0 ) {
546                   debug_handler( "This class was freed by a recursive call, exiting 'remove_server_class'");
547                   //message_free(msg); - this will be freed by server_class_handle_msg
548                   debug_handler( "Message Freed");
549                   return -1;
550                 }
551                 message_free(msg);
552         }
553         
554
555         free( class->server_class );
556         class->server_class = NULL;
557
558         find_server_class( router, router_resource ); /* find deletes for us */
559         debug_handler( "Successfuly called 'find_server_class' on class to delete" );
560
561         if( router->server_class_list == NULL ) 
562                 return 0;
563         return 1;
564 }
565
566 server_class_node * find_server_class ( transport_router_registrar * router, const char * class_id ) {
567
568         if ( router == NULL ) {
569                 warning_handler(" find_server_class(): bad arg!");
570                 return NULL;
571         }
572
573         info_handler( "Finding server class for %s", class_id );
574         server_class_node * class = router->server_class_list;
575         server_class_node * dead_class = NULL;
576
577         while ( class != NULL ) {
578
579                 if ( class->server_class == NULL ) {
580                         info_handler( "Found an empty server class" );
581
582                         if ( class->prev != NULL ) {
583                                 class->prev->next = class->next;
584                                 if( class->next != NULL ) {
585                                         class->next->prev = class->prev;
586                                 }
587
588                         } else {
589                                 info_handler( "Empty class is the first on the list" );
590                                 if( class->next != NULL ) 
591                                         router->server_class_list = class->next;
592
593                                 else { /* we're the last class node in the class node list */
594                                         info_handler( "Empty class is the last on the list" );
595                                         server_class_node_free( router->server_class_list );
596                                         router->server_class_list = NULL;
597                                         break;
598                                 }
599                                         
600                         }
601
602                         dead_class = class;
603                         class = class->next;
604
605                         info_handler( "Tossing our dead class" );
606                         server_class_node_free( dead_class );
607
608                         if ( class == NULL )
609                                 return NULL;
610                 }
611
612                 if ( strcmp(class->server_class, class_id) == 0 )
613                         return class;
614                 info_handler( "%s != %s", class->server_class, class_id );
615
616                 class = class->next;
617         }
618
619         debug_handler( "Returning NULL from find_server_class()" );
620         return NULL;
621 }
622
623 /* builds a new server class and connects to the jabber server with the new resource */
624 server_class_node* init_server_class( 
625                 transport_router_registrar* router, char* remote_id, char* server_class ) {
626
627         size_t len = sizeof( server_class_node );
628         server_class_node* node = (server_class_node*) safe_malloc( len );
629
630         node->jabber = jabber_connect_init( router->jabber->server,
631                         router->jabber->port, router->jabber->unixpath, router->jabber->username, 
632                         router->jabber->password, server_class, router->jabber->connect_timeout, router->component );
633
634
635
636         node->server_class = strdup( server_class );
637         if( server_class == NULL ) {
638                 fatal_handler( "imit_server_class(): out of memory for %s", server_class );
639                 return NULL;
640         }
641
642         info_handler( "Received class to init_server_class: %s", server_class );
643         node->current_server_node = init_server_node( remote_id );
644         if( node->current_server_node == NULL ) {
645                 fatal_handler( "init_server_class(): NULL server_node for %s", remote_id );
646                 return NULL;
647         }
648
649
650         if( ! j_connect( node->jabber ) ) {
651                 fatal_handler( "Unable to init server class %s", node->server_class );
652                 return NULL;
653         }
654
655         info_handler( "Jabber address in init for %s : address %x : username %s : resource %s", 
656                         node->server_class, node->jabber->t_client->session->sock_id, 
657                         node->jabber->username,  node->jabber->resource );
658
659         return node;
660
661 }
662
663 /* builds a new server_node to be added to the ring of server_nodes */
664 server_node* init_server_node(  char* remote_id ) {
665
666         info_handler( "Initing server node for %s", remote_id );
667         server_node* current_server_node;
668         size_t size = sizeof( server_node);
669         current_server_node = (server_node*) safe_malloc( size );
670
671         current_server_node->remote_id = strdup(remote_id);
672         if( current_server_node->remote_id == NULL ) {
673                 fatal_handler("init_server_class(): Out of Memory for %s", remote_id );
674                 return NULL;
675         }
676         
677         current_server_node->reg_time = time(NULL);     
678         current_server_node->available = 1;
679         current_server_node->next = current_server_node;
680         current_server_node->prev = current_server_node;
681
682
683         return current_server_node;
684
685 }
686
687 int  server_class_handle_msg( transport_router_registrar* router, 
688                 server_class_node* s_node, transport_message* msg ) {
689
690         if( s_node->current_server_node == NULL ) {
691                 /* return error to client ??!*/
692                 /* WE have no one to send the message to */
693                 warning_handler( "We no longer have any servers for %s : " 
694                                 "no one to send the message to. Sending error message to %s", s_node->server_class, msg->sender );
695                 free( msg->recipient );  
696
697                 char* rec = strdup( msg->sender );
698                 if( rec == NULL ) {
699                         fatal_handler( "class msg_handler: out of memory");
700                         return 0;
701                 }
702
703                 info_handler( "Building error message to return for %s", s_node->server_class);
704                 msg->recipient = rec;
705                 set_msg_error(msg, "cancel", 501);
706
707                 client_send_message( s_node->jabber->t_client, msg );
708                 message_free( msg );
709
710                 remove_server_class( router, s_node );
711                 debug_handler( "Successfully removed server class" ); 
712
713                 return -1;
714         }
715
716         info_handler( "[%s] Received \nfrom: %s \nto: %s", 
717                         s_node->server_class, msg->sender, msg->recipient );
718
719         if( msg->is_error ) {
720                 warning_handler( "We've received an error message type: %s : code: %d", 
721                                 msg->error_type, msg->error_code );
722
723                 if( strcmp( msg->error_type, "cancel" ) == 0 ) {
724                         warning_handler( "Looks like we've lost a server!" );
725                         server_node* dead_node = find_server_node( s_node, msg->sender );
726
727                         if( dead_node != NULL ) { 
728                                 //message_free( msg );
729                                 transport_message* tmp = dead_node->last_sent;
730
731                                 /* copy over last sent, it will be freed in the unregister function */
732                                 transport_message* tmp2 = message_init( tmp->body, tmp->subject, tmp->thread,
733                                                 tmp->recipient, tmp->sender );
734                                         
735                                 message_set_router_info( tmp2, tmp->router_from,  
736                                                 tmp->router_to, tmp->router_class, tmp->router_command, tmp->broadcast );
737
738                                 if( ! unregister_server_node( s_node, dead_node->remote_id ) ) { 
739                                         /* WE have no one to send the message to */
740                                         warning_handler( "We no longer have any servers for %s : " 
741                                                         "no one to send the message to.", s_node->server_class );
742                                         free( msg->recipient );  
743
744                                         char* rec = strdup( msg->router_from );
745                                         if( rec == NULL ) {
746                                                 fatal_handler( "class msg_handler: out of memory");
747                                                 return 0;
748                                         }
749
750                                         info_handler( "Building error message to return for %s", s_node->server_class);
751                                         msg->recipient = rec;
752                                         client_send_message( s_node->jabber->t_client, msg );
753                                         message_free( tmp2 );
754                                         message_free( msg );
755                                         return 0;
756
757                                 } else {
758                                         msg = tmp2;
759                                 }
760                         }
761                 }
762         } 
763
764
765         server_node* c_node = s_node->current_server_node->next;
766
767         /* not implemented yet */
768         while( ! c_node->available ) {
769                 if( c_node == s_node->current_server_node ) {
770                         warning_handler("No server_node's are available for %s", s_node->server_class );
771                         /* XXX send error message to client */
772                         return 0;
773                 }
774                 c_node = c_node->next;
775         }
776         s_node->current_server_node = c_node;
777
778         transport_message * new_msg =
779                 message_init(   msg->body, msg->subject, msg->thread, 
780                                 s_node->current_server_node->remote_id, msg->sender );
781
782         message_set_router_info( new_msg, msg->sender, NULL, NULL, NULL, 0 );
783
784         info_handler( "[%s] Routing message from [%s]\nto [%s]", s_node->server_class, msg->sender, new_msg->recipient );
785         //info_handler( "New Message Details: sender:%s recipient: %s \nbody: %s", 
786         //              new_msg->sender, new_msg->recipient, new_msg->body );
787
788         message_free( s_node->current_server_node->last_sent );
789         s_node->current_server_node->last_sent = msg;
790
791         if ( new_msg != NULL && client_send_message( s_node->jabber->t_client, new_msg ) ) {
792                 s_node->current_server_node->serve_count++;
793                 s_node->current_server_node->la_time = time(NULL);
794                 message_free( new_msg ); // XXX
795                 return 1;
796         }
797         message_free( new_msg ); // XXX
798
799         return 0;
800 }
801
802
803 int router_return_server_info( 
804                 transport_router_registrar* router, transport_message* msg ) {
805
806         server_class_node* cur_class = router->server_class_list;
807         growing_buffer* buffer = buffer_init(1024);
808
809         while( cur_class != NULL ) {
810
811                 server_node* start_node = cur_class->current_server_node;
812                 server_node* cur_node = start_node;
813                 if( cur_node == NULL ) continue; 
814
815                 do {
816
817                         char tbuf[124];
818                         memset(tbuf,0,124);
819                         sprintf( tbuf, "%d", (int)cur_node->reg_time ); 
820                         buffer_add( buffer, tbuf );
821                         buffer_add( buffer, " | ");
822
823                         memset(tbuf,0,124);
824                         sprintf( tbuf, "%d", (int)cur_node->upd_time ); 
825                         buffer_add( buffer, tbuf );
826                         buffer_add( buffer, " | ");
827
828                         memset(tbuf,0,124);
829                         sprintf( tbuf, "%d", (int)cur_node->la_time );  
830                         buffer_add( buffer, tbuf );
831                         buffer_add( buffer, " | ");
832
833
834                         char sbuf[64];
835                         memset(sbuf,0,64);
836                         sprintf(sbuf,"%d",cur_node->serve_count);
837
838                         buffer_add( buffer, "#" ); 
839                         buffer_add( buffer, sbuf ); 
840                         buffer_add( buffer, " | ");
841
842                         buffer_add( buffer, cur_class->server_class );
843                         buffer_add( buffer, " | ");
844
845                         buffer_add( buffer, cur_node->remote_id );
846
847                         buffer_add( buffer, "\n" );
848                         cur_node = cur_node->next;
849
850                 } while( cur_node != start_node );
851
852                 cur_class = cur_class->next;
853
854         }
855
856         info_handler( "Router returning servers query command: %s", buffer->buf );
857
858         transport_message* new_msg; 
859
860         if( buffer->buf == NULL || strlen(buffer->buf) == 0 )
861                 new_msg = message_init( "0", NULL, NULL, msg->sender, NULL );
862         else
863                 new_msg = message_init( buffer->buf, NULL, NULL, msg->sender, NULL );
864
865         client_send_message( router->jabber->t_client, new_msg );
866         message_free( new_msg );
867
868         return 1;
869
870 }
871
872 int router_registrar_handle_app_request( 
873                 transport_router_registrar* router, transport_message* msg ) {
874
875         osrf_message* arr[ROUTER_MAX_MSGS_PER_PACKET];
876         memset(arr, 0, ROUTER_MAX_MSGS_PER_PACKET );
877         //int num_msgs = osrf_message_from_xml( msg->body, arr );
878         int num_msgs = osrf_message_deserialize( msg->body, arr, ROUTER_MAX_MSGS_PER_PACKET );
879
880         int i;
881         for( i = 0; i != num_msgs; i++ ) {
882
883                 osrf_message* omsg = arr[i];
884                 osrf_message* success = NULL;
885                 osrf_message** result = NULL;
886                 int num_responses;
887
888                 //char* newxml =  osrf_message_to_xml(omsg);
889                 //debug_handler( "Received potential app request from client:\n%s\n", newxml ); 
890                 //free(newxml);
891
892                 if(omsg->m_type == CONNECT) {
893
894                         success = osrf_message_init( 
895                                         STATUS, omsg->thread_trace, omsg->protocol );
896
897                         osrf_message_set_status_info( 
898                                         success, "osrfConnectStatus", "Connection Successful", OSRF_STATUS_OK );
899
900                 } else if( omsg->m_type == REQUEST ) {
901
902                         result = router_registrar_process_app_request( router, omsg, &num_responses );
903                         
904                 } else if(omsg->m_type == DISCONNECT) { 
905                         success = NULL;
906
907                 } else {
908
909                         success = osrf_message_init( 
910                                         STATUS, omsg->thread_trace, omsg->protocol );
911                         osrf_message_set_status_info( 
912                                         success, "osrfMethodException", "Method Not Found", OSRF_STATUS_NOTFOUND );
913                 }
914
915
916                 /* now send our new message back */
917                 if(success) {
918
919                         char* xml       = osrf_message_serialize(success);
920                         debug_handler( "Sending XML to client app request:\n%s\n", xml );
921                         transport_message* return_m = message_init( 
922                                 xml, "", msg->thread, msg->sender, "" );
923
924
925                         client_send_message(router->jabber->t_client, return_m);
926
927                         free(xml);
928                         osrf_message_free(success);
929                 } else if(result) {
930                         int i;
931
932                         for(i=0; i!= num_responses; i++){
933                                 char* xml =  osrf_message_serialize(result[i]);
934                                 debug_handler( "Sending XML to client app request:\n%s\n", xml );
935                                 transport_message* return_m = message_init( 
936                                         xml, "", msg->thread, msg->sender, "" );
937
938
939                                 client_send_message(router->jabber->t_client, return_m);
940
941                                 free(xml);
942                                 osrf_message_free(result[i]);
943                         }
944
945                         osrf_message* complete = osrf_message_init(
946                                 STATUS, result[i-1]->thread_trace, result[i-1]->protocol );
947                         osrf_message_set_status_info( complete, 
948                                         "osrfConnectStatus", "Request Complete", OSRF_STATUS_COMPLETE );
949                         char* complete_xml = osrf_message_serialize(complete);
950                         transport_message* complete_m = message_init( 
951                                         complete_xml, "", msg->thread, msg->sender, "" );
952                         client_send_message(router->jabber->t_client, complete_m);
953                         free(complete_xml);
954                         osrf_message_free(complete);
955                         free(result);
956
957                 }
958
959         }
960
961         return 1;
962
963 }
964
965
966
967
968 osrf_message** router_registrar_process_app_request( 
969                 transport_router_registrar* router, osrf_message* omsg, int* num_responses ) {
970
971
972         if( omsg->method_name == NULL )
973                 return NULL;
974
975         osrf_message** result_array = NULL;
976
977
978         debug_handler( "Received method from client: %s", omsg->method_name );
979
980         if(!strcmp(omsg->method_name,"opensrf.router.info.class.list")) {
981
982                 jsonObject* result_content = jsonParseString("[]");
983
984                 debug_handler("Processing opensrf.router.info.class.list request");
985
986                 server_class_node* cur_class = router->server_class_list;
987                 while( cur_class != NULL ) {
988
989                         debug_handler("Adding %s to request list", cur_class->server_class);
990
991                         jsonObjectPush(result_content, jsonNewObject(cur_class->server_class));
992                         cur_class = cur_class->next;
993                 }
994                 result_array = safe_malloc(sizeof(osrf_message*));
995                 *num_responses = 1;
996
997                 result_array[0] = osrf_message_init(
998                         RESULT, omsg->thread_trace, omsg->protocol );
999
1000                 osrf_message_set_result_content( result_array[0], jsonObjectToJSON(result_content));
1001                 jsonObjectFree(result_content);
1002
1003
1004         } else if(!strcmp(omsg->method_name,"opensrf.router.info.stats")) {
1005
1006                 /* we have a server_class array which holds an array of node_objects.
1007                         each node_object contains an array of stats
1008                         */
1009
1010                 debug_handler("Processing opensrf.router.info.stats request");
1011
1012                 jsonObject* result_content = jsonParseString("{}");
1013
1014                 server_class_node* cur_class = router->server_class_list;
1015
1016                 while( cur_class != NULL ) {
1017
1018                         server_node* start_node = cur_class->current_server_node;
1019                         server_node* cur_node = start_node;
1020                         if( cur_node == NULL ) continue; 
1021
1022                         jsonObject* server_object = jsonParseString("{}");
1023
1024                         do {
1025
1026                                 jsonObject* node_stats_array = jsonParseString("[]");
1027
1028                                 jsonObject* json_reg_time = jsonParseString("{}");
1029
1030                                 jsonObjectSetKey( json_reg_time, "reg_time", 
1031                                         jsonNewNumberObject((double) cur_node->reg_time));
1032
1033                                 jsonObjectPush(  node_stats_array, json_reg_time );
1034
1035                                 jsonObject* json_upd_time = jsonParseString("{}");
1036
1037
1038                                 jsonObjectSetKey( json_upd_time, "upd_time", 
1039                                         jsonNewNumberObject((int)cur_node->upd_time));
1040
1041
1042
1043                                 jsonObjectPush( node_stats_array, json_upd_time );
1044
1045
1046
1047                                 jsonObject* json_la_time = jsonParseString("{}");
1048
1049
1050
1051                                 jsonObjectSetKey( json_la_time, "la_time", 
1052                                                 jsonNewNumberObject((int)cur_node->la_time));
1053
1054
1055                                 jsonObjectPush( node_stats_array, json_la_time );
1056
1057                                 jsonObject* json_serve_count = jsonParseString("{}");
1058
1059
1060                                 jsonObjectSetKey( json_serve_count, "serve_count", 
1061                                         jsonNewNumberObject((int)cur_node->serve_count));
1062
1063                                 
1064                                 jsonObjectPush( node_stats_array, json_serve_count );
1065
1066
1067                                 jsonObjectSetKey( server_object, cur_node->remote_id, node_stats_array );
1068
1069                                 cur_node = cur_node->next;
1070         
1071                         } while( cur_node != start_node );
1072
1073                         jsonObjectSetKey( result_content, cur_class->server_class, server_object );
1074         
1075                         cur_class = cur_class->next;
1076
1077                                 
1078                 }
1079
1080                 result_array = safe_malloc(sizeof(osrf_message*));
1081                 *num_responses = 1;
1082
1083                 result_array[0] = osrf_message_init(
1084                         RESULT, omsg->thread_trace, omsg->protocol );
1085
1086                 osrf_message_set_result_content(result_array[0], jsonObjectToJSON(result_content));
1087
1088                 jsonObjectFree(result_content);
1089
1090
1091         } else if(!strcmp(omsg->method_name,"opensrf.system.method.all")) {
1092
1093                 jsonObject* content = jsonParseString("{}");
1094                 jsonObjectSetKey(content, "api_level", jsonNewObject("1"));
1095                 jsonObjectSetKey(content, "api_name", jsonNewObject("opensrf.router.info.class.list"));
1096                 jsonObjectSetKey(content, "server_class", jsonNewObject("router"));
1097                 jsonObjectSetKey(content, "stream", jsonNewObject("0"));
1098
1099                 jsonObject* content2 = jsonParseString("{}");
1100                 jsonObjectSetKey(content2, "api_level", jsonNewObject("1"));
1101                 jsonObjectSetKey(content2, "api_name", jsonNewObject("opensrf.router.info.stats"));
1102                 jsonObjectSetKey(content2, "server_class", jsonNewObject("router"));
1103                 jsonObjectSetKey(content2, "stream", jsonNewObject("0"));
1104
1105                 jsonObject* content3 = jsonParseString("{}");
1106                 jsonObjectSetKey(content3, "api_level", jsonNewObject("1"));
1107                 jsonObjectSetKey(content3, "api_name", jsonNewObject("opensrf.system.method.all"));
1108                 jsonObjectSetKey(content3, "server_class", jsonNewObject("router"));
1109                 jsonObjectSetKey(content3, "stream", jsonNewObject("1"));
1110
1111                 result_array = safe_malloc(3*sizeof(osrf_message*));
1112                 *num_responses = 3;
1113
1114                 result_array[0] = osrf_message_init(
1115                         RESULT, omsg->thread_trace, omsg->protocol );
1116
1117                 osrf_message_set_result_content( result_array[0], jsonObjectToJSON(content));
1118                 jsonObjectFree(content);
1119
1120                 result_array[1] = osrf_message_init(
1121                         RESULT, omsg->thread_trace, omsg->protocol );
1122                 osrf_message_set_result_content( result_array[1], jsonObjectToJSON(content2) );
1123                 jsonObjectFree(content2);
1124
1125                 result_array[2] = osrf_message_init(
1126                         RESULT, omsg->thread_trace, omsg->protocol );
1127                 osrf_message_set_result_content( result_array[1], jsonObjectToJSON(content3) );
1128                 jsonObjectFree(content3);
1129
1130
1131         }
1132         
1133         
1134         if( result_array == NULL || result_array[0] == NULL ) 
1135                 return NULL;
1136
1137         return result_array;
1138 }
1139
1140
1141
1142
1143 int router_registrar_free( transport_router_registrar* router_registrar ) {
1144         if( router_registrar == NULL ) return 0;
1145         jabber_connect_free( router_registrar->jabber );
1146
1147         /* free the server_class list XXX */
1148         while( router_registrar->server_class_list != NULL ) {
1149                 remove_server_class(router_registrar, router_registrar->server_class_list);
1150                 debug_handler( "Removed server classes in registrar free");
1151         }
1152
1153         osrfStringArrayFree(router_registrar->trusted_servers);
1154         osrfStringArrayFree(router_registrar->trusted_clients);
1155
1156         free( router_registrar );
1157         return 1;
1158 }
1159
1160
1161 int server_class_node_free( server_class_node* node ) {
1162         if( node == NULL ) { return 0; }
1163         if( node->server_class != NULL ) 
1164                 free( node->server_class );
1165
1166         jabber_connect_free( node->jabber );
1167
1168         /* just in case, free the list */
1169         while( node->current_server_node != NULL ) {
1170                 unregister_server_node( node, node->current_server_node->remote_id );
1171         }
1172         free( node );
1173         return 1;
1174 }
1175
1176 int server_node_free( server_node* node ) {
1177         if( node == NULL ) { return 0; }
1178         message_free( node->last_sent );
1179         free( node->remote_id );
1180         free( node );
1181         return 1;
1182 }
1183
1184 int jabber_connect_free( jabber_connect* jabber ) {
1185         if( jabber == NULL ) { return 0; }
1186         client_free( jabber->t_client );
1187         free( jabber->unixpath );
1188         free( jabber->username );
1189         free( jabber->password );
1190         free( jabber->resource );
1191         free( jabber->server );
1192         free( jabber );
1193         return 1;
1194 }
1195
1196