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