src/router/osrf_router.c

   1 #include <sys/select.h>
   2 #include <signal.h>
   3 #include "opensrf/utils.h"
   4 #include "opensrf/log.h"
   5 #include "opensrf/osrf_list.h"
   6 #include "opensrf/string_array.h"
   7 #include "opensrf/osrf_hash.h"
   8 #include "osrf_router.h"
   9 #include "opensrf/transport_client.h"
  10 #include "opensrf/transport_message.h"
  11 #include "opensrf/osrf_message.h"
  12
  13 /**
  14         @file osrf_router.c
  15         @brief Implementation of osrfRouter.
  16
  17         The router opens multiple Jabber sessions for the same username and domain, one for
  18         each server class.  The Jabber IDs for these sessions are distinguished by the use of
  19         the class names as Jabber resource names.
  20
  21         For each server class there may be multiple server nodes.  Each node corresponds to a
  22         listener process for a service.
  23 */
  24
  25 /**
  26         @brief Collection of server classes, with connection parameters for Jabber.
  27  */
  28 struct osrfRouterStruct {
  29
  30         /**
  31                 @brief Hash store of server classes.
  32
  33                 For each entry, the key is the class name, and the corresponding datum is an
  34                 osrfRouterClass.
  35         */
  36         osrfHash* classes;
  37         osrfHashIterator* class_itr;  /**< For traversing the list of classes. */
  38         char* domain;         /**< Domain name of Jabber server. */
  39         char* name;           /**< Router's username for the Jabber logon. */
  40         char* resource;       /**< Router's resource name for the Jabber logon. */
  41         char* password;       /**< Router's password for the Jabber logon. */
  42         int port;             /**< Jabber's port number. */
  43         sig_atomic_t stop;    /**< To be set by signal handler to interrupt main loop. */
  44
  45         /** Array of client domains that we allow to send requests through us. */
  46         osrfStringArray* trustedClients;
  47         /** Array of server domains that we allow to register, etc. with us. */
  48         osrfStringArray* trustedServers;
  49
  50         transport_client* connection;
  51 };
  52
  53 /**
  54         @brief Maintains a set of server nodes belonging to the same class.
  55 */
  56 struct _osrfRouterClassStruct {
  57         osrfRouter* router;         /**< The osrfRouter that owns this osrfRouterClass. */
  58         osrfHashIterator* itr;      /**< Iterator for set of osrfRouterNodes. */
  59         /**
  60                 @brief Hash store of server nodes.
  61
  62                 The key of each entry is a node name, and the associated data is an osrfRouterNode.
  63                 We install a callback for freeing the entries.
  64         */
  65         osrfHash* nodes;
  66         /** The transport_client used for communicating with this server. */
  67         transport_client* connection;
  68 };
  69 typedef struct _osrfRouterClassStruct osrfRouterClass;
  70
  71 /**
  72         @brief Represents a link to a single server's inbound connection.
  73 */
  74 struct _osrfRouterNodeStruct {
  75         char* remoteId;     /**< Send message to me via this login. */
  76         int count;          /**< How many message have been sent to this node. */
  77         transport_message* lastMessage;
  78 };
  79 typedef struct _osrfRouterNodeStruct osrfRouterNode;
  80
  81 static osrfRouterClass* osrfRouterAddClass( osrfRouter* router, const char* classname );
  82 static void osrfRouterClassAddNode( osrfRouterClass* rclass, const char* remoteId );
  83 static void osrfRouterHandleCommand( osrfRouter* router, const transport_message* msg );
  84 static void osrfRouterClassHandleMessage( osrfRouter* router,
  85                 osrfRouterClass* rclass, const transport_message* msg );
  86 static void osrfRouterRemoveClass( osrfRouter* router, const char* classname );
  87 static void osrfRouterClassRemoveNode( osrfRouter* router, const char* classname,
  88                 const char* remoteId );
  89 static void osrfRouterClassFree( char* classname, void* rclass );
  90 static void osrfRouterNodeFree( char* remoteId, void* node );
  91 static osrfRouterClass* osrfRouterFindClass( osrfRouter* router, const char* classname );
  92 static osrfRouterNode* osrfRouterClassFindNode( osrfRouterClass* rclass,
  93                 const char* remoteId );
  94 static int _osrfRouterFillFDSet( osrfRouter* router, fd_set* set );
  95 static void osrfRouterHandleIncoming( osrfRouter* router );
  96 static void osrfRouterClassHandleIncoming( osrfRouter* router,
  97                 const char* classname,  osrfRouterClass* class );
  98 static transport_message* osrfRouterClassHandleBounce( osrfRouter* router,
  99                 const char* classname, osrfRouterClass* rclass, const transport_message* msg );
 100 static void osrfRouterHandleAppRequest( osrfRouter* router, const transport_message* msg );
 101 static void osrfRouterRespondConnect( osrfRouter* router, const transport_message* msg,
 102                 const osrfMessage* omsg );
 103 static void osrfRouterProcessAppRequest( osrfRouter* router, const transport_message* msg,
 104                 const osrfMessage* omsg );
 105 static void osrfRouterSendAppResponse( osrfRouter* router, const transport_message* msg,
 106                 const osrfMessage* omsg, const jsonObject* response );
 107 static void osrfRouterHandleMethodNFound( osrfRouter* router,
 108                 const transport_message* msg, const osrfMessage* omsg );
 109
 110 #define ROUTER_REGISTER "register"
 111 #define ROUTER_UNREGISTER "unregister"
 112
 113 #define ROUTER_REQUEST_CLASS_LIST "opensrf.router.info.class.list"
 114 #define ROUTER_REQUEST_STATS_NODE_FULL "opensrf.router.info.stats.class.node.all"
 115 #define ROUTER_REQUEST_STATS_CLASS_FULL "opensrf.router.info.stats.class.all"
 116 #define ROUTER_REQUEST_STATS_CLASS "opensrf.router.info.stats.class"
 117 #define ROUTER_REQUEST_STATS_CLASS_SUMMARY "opensrf.router.info.stats.class.summary"
 118
 119 /**
 120         @brief Stop the otherwise endless main loop of the router.
 121         @param router Pointer to the osrfRouter to be stopped.
 122
 123         To be called by a signal handler.  We don't stop the loop immediately; we just set
 124         a switch that the main loop checks on each iteration.
 125 */
 126 void router_stop( osrfRouter* router )
 127 {
 128         if( router )
 129                 router->stop = 1;
 130 }
 131
 132 /**
 133         @brief Allocate and initialize a new osrfRouter.
 134         @param domain Domain name of Jabber server.
 135         @param name Router's username for the Jabber logon.
 136         @param resource Router's resource name for the Jabber logon.
 137         @param password Router's password for the Jabber logon.
 138         @param port Jabber's port number.
 139         @param trustedClients Array of client domains that we allow to send requests through us.
 140         @param trustedServers Array of server domains that we allow to register, etc. with us.
 141         @return Pointer to the newly allocated osrfRouter, or NULL upon error.
 142
 143         Don't connect to Jabber yet.  We'll do that later, upon a call to osrfRouterConnect().
 144
 145         The calling code is responsible for freeing the osrfRouter by calling osrfRouterFree().
 146 */
 147 osrfRouter* osrfNewRouter(
 148                 const char* domain, const char* name,
 149                 const char* resource, const char* password, int port,
 150                 osrfStringArray* trustedClients, osrfStringArray* trustedServers ) {
 151
 152         if(!( domain && name && resource && password && port && trustedClients && trustedServers ))
 153                 return NULL;
 154
 155         osrfRouter* router     = safe_malloc(sizeof(osrfRouter));
 156         router->domain         = strdup(domain);
 157         router->name           = strdup(name);
 158         router->password       = strdup(password);
 159         router->resource       = strdup(resource);
 160         router->port           = port;
 161         router->stop           = 0;
 162
 163         router->trustedClients = trustedClients;
 164         router->trustedServers = trustedServers;
 165
 166
 167         router->classes = osrfNewHash();
 168         osrfHashSetCallback(router->classes, &osrfRouterClassFree);
 169         router->class_itr = osrfNewHashIterator( router->classes );
 170
 171         // Prepare to connect to Jabber, as a non-component, over TCP (not UNIX domain).
 172         router->connection = client_init( domain, port, NULL, 0 );
 173
 174         return router;
 175 }
 176
 177 /**
 178         @brief Connect to Jabber.
 179         @param router Pointer to the osrfRouter to connect to Jabber.
 180         @return 0 if successful, or -1 on error.
 181
 182         Allow up to 10 seconds for the logon to succeed.
 183
 184         We connect over TCP (not over a UNIX domain), as a non-component.
 185 */
 186 int osrfRouterConnect( osrfRouter* router ) {
 187         if(!router) return -1;
 188         int ret = client_connect( router->connection, router->name,
 189                         router->password, router->resource, 10, AUTH_DIGEST );
 190         if( ret == 0 ) return -1;
 191         return 0;
 192 }
 193
 194 /**
 195         @brief Enter endless loop to receive and respond to input.
 196         @param router Pointer to the osrfRouter that's looping.
 197
 198         On each iteration: wait for incoming messages to arrive on any of our sockets -- i.e.
 199         either the top level socket belonging to the router or any of the lower level sockets
 200         belonging to the classes.  React to the incoming activity as needed.
 201
 202         We don't exit the loop until we receive a signal to stop, or until we encounter an error.
 203 */
 204 void osrfRouterRun( osrfRouter* router ) {
 205         if(!(router && router->classes)) return;
 206
 207         int routerfd = client_sock_fd( router->connection );
 208         int selectret = 0;
 209
 210         // Loop until a signal handler sets router->stop
 211         while( ! router->stop ) {
 212
 213                 fd_set set;
 214                 int maxfd = _osrfRouterFillFDSet( router, &set );
 215
 216                 // Wait indefinitely for an incoming message
 217                 if( (selectret = select(maxfd + 1, &set, NULL, NULL, NULL)) < 0 ) {
 218                         if( EINTR == errno ) {
 219                                 if( router->stop ) {
 220                                         osrfLogWarning( OSRF_LOG_MARK, "Top level select call interrupted by signal" );
 221                                         break;
 222                                 }
 223                                 else
 224                                         continue;    // Irrelevant signal; ignore it
 225                         } else {
 226                                 osrfLogWarning( OSRF_LOG_MARK, "Top level select call failed with errno %d: %s",
 227                                                 errno, strerror( errno ) );
 228                                 break;
 229                         }
 230                 }
 231
 232                 /* see if there is a top level router message */
 233                 if( FD_ISSET(routerfd, &set) ) {
 234                         osrfLogDebug( OSRF_LOG_MARK, "Top router socket is active: %d", routerfd );
 235                         osrfRouterHandleIncoming( router );
 236                 }
 237
 238                 /* Check each of the connected classes and see if they have data to route */
 239                 osrfRouterClass* class;
 240                 osrfHashIterator* itr = router->class_itr;  // remove a layer of indirection
 241                 osrfHashIteratorReset( itr );
 242
 243                 while( (class = osrfHashIteratorNext(itr)) ) {   // for each class
 244
 245                         const char* classname = osrfHashIteratorKey(itr);
 246
 247                         if( classname ) {
 248
 249                                 osrfLogDebug( OSRF_LOG_MARK, "Checking %s for activity...", classname );
 250
 251                                 int sockfd = client_sock_fd( class->connection );
 252                                 if(FD_ISSET( sockfd, &set )) {
 253                                         osrfLogDebug( OSRF_LOG_MARK, "Socket is active: %d", sockfd );
 254                                         osrfRouterClassHandleIncoming( router, classname, class );
 255                                 }
 256                         }
 257                 } // end while
 258         } // end while
 259 }
 260
 261
 262 /**
 263         @brief Handle incoming requests to the router.
 264         @param router Pointer to the osrfRouter.
 265
 266         Read all available input messages from the top-level transport_client.  For each one:
 267         if the domain of the sender's Jabber id is on the list of approved domains, pass the
 268         message to osrfRouterHandleCommand() (if there's a message) or osrfRouterHandleAppRequest()
 269         (if there isn't).  If the domain is @em not on the approved list, log a warning and
 270         discard the message.
 271 */
 272 static void osrfRouterHandleIncoming( osrfRouter* router ) {
 273         if(!router) return;
 274
 275         transport_message* msg = NULL;
 276
 277         while( (msg = client_recv( router->connection, 0 )) ) {  // for each message
 278
 279                 if( msg->sender ) {
 280
 281                         osrfLogDebug(OSRF_LOG_MARK,
 282                                 "osrfRouterHandleIncoming(): investigating message from %s", msg->sender);
 283
 284                         /* if the server is not on a trusted domain, drop the message */
 285                         int len = strlen(msg->sender) + 1;
 286                         char domain[len];
 287                         jid_get_domain( msg->sender, domain, len - 1 );
 288
 289                         if(osrfStringArrayContains( router->trustedServers, domain)) {
 290
 291                                 // If there's a command, obey it.  Otherwise, treat
 292                                 // the message as an app session level request.
 293                                 if( msg->router_command && *msg->router_command )
 294                                         osrfRouterHandleCommand( router, msg );
 295                                 else
 296                                         osrfRouterHandleAppRequest( router, msg );
 297                         }
 298                         else
 299                                 osrfLogWarning( OSRF_LOG_MARK,
 300                                                 "Received message from un-trusted server domain %s", msg->sender);
 301                 }
 302
 303                 message_free(msg);
 304         }
 305 }
 306
 307 /**
 308         @brief Handle all available incoming messages for a router class.
 309         @param router Pointer to the osrfRouter.
 310         @param classname Class name.
 311         @param class Pointer to the osrfRouterClass.
 312
 313         For each message: if the sender is on a trusted domain, process the message.
 314 */
 315 static void osrfRouterClassHandleIncoming( osrfRouter* router, const char* classname,
 316                 osrfRouterClass* class ) {
 317         if(!(router && class)) return;
 318
 319         transport_message* msg;
 320         osrfLogDebug( OSRF_LOG_MARK, "osrfRouterClassHandleIncoming()");
 321
 322         // For each incoming message for this class:
 323         while( (msg = client_recv( class->connection, 0 )) ) {
 324
 325                 // Save the transaction id so that we can incorporate it
 326                 // into any relevant messages
 327                 osrfLogSetXid(msg->osrf_xid);
 328
 329                 if( msg->sender ) {
 330
 331                         osrfLogDebug(OSRF_LOG_MARK,
 332                                 "osrfRouterClassHandleIncoming(): investigating message from %s", msg->sender);
 333
 334                         /* if the client is not from a trusted domain, drop the message */
 335                         int len = strlen(msg->sender) + 1;
 336                         char domain[len];
 337                         jid_get_domain( msg->sender, domain, len - 1 );
 338
 339                         if(osrfStringArrayContains( router->trustedClients, domain )) {
 340
 341                                 if( msg->is_error )  {
 342
 343                                         // A previous message bounced.  Try to send a clone of it to a
 344                                         // different node of the same class.
 345                                         transport_message* bouncedMessage = osrfRouterClassHandleBounce(
 346                                                         router, classname, class, msg );
 347                                         /* handle bounced message */
 348                                         if( !bouncedMessage ) {
 349                                                 /* we have no one to send the requested message to */
 350                                                 message_free( msg );
 351                                                 osrfLogClearXid();
 352                                                 continue;
 353                                         }
 354                                         osrfRouterClassHandleMessage( router, class, bouncedMessage );
 355                                         message_free( bouncedMessage );
 356                                 } else
 357                                         osrfRouterClassHandleMessage( router, class, msg );
 358
 359                         } else {
 360                                 osrfLogWarning( OSRF_LOG_MARK,
 361                                                 "Received client message from untrusted client domain %s", domain );
 362                         }
 363                 }
 364
 365                 message_free( msg );
 366                 osrfLogClearXid();  // We're done with this transaction id
 367         }
 368 }
 369
 370 /**
 371         @brief Handle a top level router command.
 372         @param router Pointer to the osrfRouter.
 373         @param msg Pointer to the transport_message to be handled.
 374
 375         Currently supported commands:
 376         - "register" -- Add a server class and/or a server node to our lists.
 377         - "unregister" -- Remove a node from a class, and the class as well if no nodes are
 378         left for it.
 379 */
 380 static void osrfRouterHandleCommand( osrfRouter* router, const transport_message* msg ) {
 381         if(!(router && msg && msg->router_class)) return;
 382
 383         if( !strcmp( msg->router_command, ROUTER_REGISTER ) ) {
 384
 385                 osrfLogInfo( OSRF_LOG_MARK, "Registering class %s", msg->router_class );
 386
 387                 // Add the server class to the list, if it isn't already there
 388                 osrfRouterClass* class = osrfRouterFindClass( router, msg->router_class );
 389                 if(!class)
 390                         class = osrfRouterAddClass( router, msg->router_class );
 391
 392                 // Add the node to the osrfRouterClass's list, if it isn't already there
 393                 if(class && ! osrfRouterClassFindNode( class, msg->sender ) )
 394                         osrfRouterClassAddNode( class, msg->sender );
 395
 396         } else if( !strcmp( msg->router_command, ROUTER_UNREGISTER ) ) {
 397
 398                 if( msg->router_class && *msg->router_class ) {
 399                         osrfLogInfo( OSRF_LOG_MARK, "Unregistering router class %s", msg->router_class );
 400                         osrfRouterClassRemoveNode( router, msg->router_class, msg->sender );
 401                 }
 402         }
 403 }
 404
 405
 406 /**
 407         @brief Add an osrfRouterClass to a router, and open a connection for it.
 408         @param router Pointer to the osrfRouter.
 409         @param classname The name of the class this node handles.
 410         @return A pointer to the new osrfRouterClass, or NULL upon error.
 411
 412         Open a Jabber session to be used for this server class.  The Jabber ID incorporates the
 413         class name as the resource name.
 414 */
 415 static osrfRouterClass* osrfRouterAddClass( osrfRouter* router, const char* classname ) {
 416         if(!(router && router->classes && classname)) return NULL;
 417
 418         osrfRouterClass* class = safe_malloc(sizeof(osrfRouterClass));
 419         class->nodes = osrfNewHash();
 420         class->itr = osrfNewHashIterator(class->nodes);
 421         osrfHashSetCallback(class->nodes, &osrfRouterNodeFree);
 422         class->router = router;
 423
 424         class->connection = client_init( router->domain, router->port, NULL, 0 );
 425
 426         if(!client_connect( class->connection, router->name,
 427                         router->password, classname, 10, AUTH_DIGEST ) ) {
 428                 // Cast away the constness of classname.  Though ugly, this
 429                 // cast is benign because osrfRouterClassFree doesn't actually
 430                 // write through the pointer.  We can't readily change its
 431                 // signature because it is used for a function pointer, and
 432                 // we would have to change other signatures the same way.
 433                 osrfRouterClassFree( (char *) classname, class );
 434                 return NULL;
 435         }
 436
 437         osrfHashSet( router->classes, class, classname );
 438         return class;
 439 }
 440
 441
 442 /**
 443         @brief Add a new server node to an osrfRouterClass.
 444         @param rclass Pointer to the osrfRouterClass to which we are to add the node.
 445         @param remoteId The remote login of the osrfRouterNode.
 446 */
 447 static void osrfRouterClassAddNode( osrfRouterClass* rclass, const char* remoteId ) {
 448         if(!(rclass && rclass->nodes && remoteId)) return;
 449
 450         osrfLogInfo( OSRF_LOG_MARK, "Adding router node for remote id %s", remoteId );
 451
 452         osrfRouterNode* node = safe_malloc(sizeof(osrfRouterNode));
 453         node->count = 0;
 454         node->lastMessage = NULL;
 455         node->remoteId = strdup(remoteId);
 456
 457         osrfHashSet( rclass->nodes, node, remoteId );
 458 }
 459
 460 /**
 461         @brief Handle an input message representing a Jabber error stanza.
 462         @param router Pointer to the current osrfRouter.
 463         @param classname Name of the class to which the error stanza was sent.
 464         @param rclass Pointer to the osrfRouterClass to which the error stanza was sent.
 465         @param msg Pointer to the transport_message representing the error stanza.
 466         @return Pointer to a newly allocated transport_message; or NULL (see remarks).
 467
 468         The presumption is that the relevant node is dead.  If another node is available for
 469         the same class, then remove the dead one, create a clone of the message to be sent
 470         elsewhere, and return a pointer to it.  If there is no other node for the same class,
 471         send a cancel message back to the sender, remove both the node and the class it belongs
 472         to, and return NULL.  If we can't even do that because the entire class is dead, log
 473         a message to that effect and return NULL.
 474 */
 475 static transport_message* osrfRouterClassHandleBounce( osrfRouter* router,
 476                 const char* classname, osrfRouterClass* rclass, const transport_message* msg ) {
 477
 478         osrfLogDebug( OSRF_LOG_MARK, "osrfRouterClassHandleBounce()");
 479
 480         osrfLogInfo( OSRF_LOG_MARK, "Received network layer error message from %s", msg->sender );
 481         osrfRouterNode* node = osrfRouterClassFindNode( rclass, msg->sender );
 482         if( ! node ) {
 483                 osrfLogInfo( OSRF_LOG_MARK,
 484                         "network error occurred after we removed the class.. ignoring");
 485                 return NULL;
 486         }
 487
 488         if( osrfHashGetCount(rclass->nodes) == 1 ) { /* the last node is dead */
 489
 490                 if( node->lastMessage ) {
 491                         osrfLogWarning( OSRF_LOG_MARK,
 492                                         "We lost the last node in the class, responding with error and removing...");
 493
 494                         transport_message* error = message_init(
 495                                 node->lastMessage->body, node->lastMessage->subject,
 496                                 node->lastMessage->thread, node->lastMessage->router_from,
 497                                 node->lastMessage->recipient );
 498                         message_set_osrf_xid(error, node->lastMessage->osrf_xid);
 499                         set_msg_error( error, "cancel", 501 );
 500
 501                         /* send the error message back to the original sender */
 502                         client_send_message( rclass->connection, error );
 503                         message_free( error );
 504                 }
 505
 506                 /* remove the dead node */
 507                 osrfRouterClassRemoveNode( router, classname, msg->sender);
 508                 return NULL;
 509
 510         } else {
 511
 512                 transport_message* lastSent = NULL;
 513
 514                 if( node->lastMessage ) {
 515                         osrfLogDebug( OSRF_LOG_MARK, "Cloning lastMessage so next node can send it");
 516                         lastSent = message_init( node->lastMessage->body,
 517                                 node->lastMessage->subject, node->lastMessage->thread, "",
 518                                 node->lastMessage->router_from );
 519                         message_set_router_info( lastSent, node->lastMessage->router_from,
 520                                 NULL, NULL, NULL, 0 );
 521                         message_set_osrf_xid( lastSent, node->lastMessage->osrf_xid );
 522                 }
 523
 524                 /* remove the dead node */
 525                 osrfRouterClassRemoveNode( router, classname, msg->sender);
 526                 return lastSent;
 527         }
 528 }
 529
 530
 531 /**
 532         @brief Forward a class-level message to a listener for the corresponding service.
 533         @param router Pointer to the current osrfRouter.
 534         @param rclass Pointer to the class to which the message is directed.
 535         @param msg Pointer to the message to be forwarded.
 536
 537         Pick a node for the specified class, and forward the message to it.
 538
 539         We use an iterator, stored with the class, to maintain a position in the class's list
 540         of nodes.  Advance the iterator to pick the next node, and if we reach the end, go
 541         back to the beginning of the list.
 542 */
 543 static void osrfRouterClassHandleMessage(
 544                 osrfRouter* router, osrfRouterClass* rclass, const transport_message* msg ) {
 545         if(!(router && rclass && msg)) return;
 546
 547         osrfLogDebug( OSRF_LOG_MARK, "osrfRouterClassHandleMessage()");
 548
 549         // Pick a node, in a round-robin.
 550         osrfRouterNode* node = osrfHashIteratorNext( rclass->itr );
 551         if(!node) {   // wrap around to the beginning of the list
 552                 osrfHashIteratorReset(rclass->itr);
 553                 node = osrfHashIteratorNext( rclass->itr );
 554         }
 555
 556         if(node) {  // should always be true -- no class without a node
 557
 558                 // Build a transport message
 559                 transport_message* new_msg = message_init( msg->body,
 560                                 msg->subject, msg->thread, node->remoteId, msg->sender );
 561                 message_set_router_info( new_msg, msg->sender, NULL, NULL, NULL, 0 );
 562                 message_set_osrf_xid( new_msg, msg->osrf_xid );
 563
 564                 osrfLogInfo( OSRF_LOG_MARK,  "Routing message:\nfrom: [%s]\nto: [%s]",
 565                                 new_msg->router_from, new_msg->recipient );
 566
 567                 // Save it for possible future reference
 568                 message_free( node->lastMessage );
 569                 node->lastMessage = new_msg;
 570
 571                 // Send it
 572                 if ( client_send_message( rclass->connection, new_msg ) == 0 )
 573                         node->count++;
 574
 575                 else {
 576                         message_prepare_xml(new_msg);
 577                         osrfLogWarning( OSRF_LOG_MARK, "Error sending message from %s to %s\n%s",
 578                                         new_msg->sender, new_msg->recipient, new_msg->msg_xml );
 579                 }
 580                 // We don't free new_msg here because we saved it as node->lastMessage.
 581         }
 582 }
 583
 584
 585 /**
 586         @brief Remove a given osrfRouterClass from an osrfRouter
 587         @param router Pointer to the osrfRouter.
 588         @param classname The name of the class to be removed.
 589
 590         Delete an osrfRouterClass from the router's list of classes.  Indirectly (via a callback
 591         function installed in the osrfHash), free the osrfRouterClass and any associated nodes.
 592 */
 593 static void osrfRouterRemoveClass( osrfRouter* router, const char* classname ) {
 594         if( router && router->classes && classname ) {
 595                 osrfLogInfo( OSRF_LOG_MARK, "Removing router class %s", classname );
 596                 osrfHashRemove( router->classes, classname );
 597         }
 598 }
 599
 600
 601 /**
 602         @brief Remove a node from a class.  If the class thereby becomes empty, remove it as well.
 603         @param router Pointer to the current osrfRouter.
 604         @param classname Class name.
 605         @param remoteId Identifier for the node to be removed.
 606 */
 607 static void osrfRouterClassRemoveNode(
 608                 osrfRouter* router, const char* classname, const char* remoteId ) {
 609
 610         if(!(router && router->classes && classname && remoteId))  // sanity check
 611                 return;
 612
 613         osrfLogInfo( OSRF_LOG_MARK, "Removing router node %s", remoteId );
 614
 615         osrfRouterClass* class = osrfRouterFindClass( router, classname );
 616         if( class ) {
 617                 osrfHashRemove( class->nodes, remoteId );
 618                 if( osrfHashGetCount(class->nodes) == 0 ) {
 619                         osrfRouterRemoveClass( router, classname );
 620                 }
 621         }
 622 }
 623
 624
 625 /**
 626         @brief Free a router class object.
 627         @param classname Class name (not used).
 628         @param c Pointer to the osrfRouterClass, cast to a void pointer.
 629
 630         This function is invoked as a callback when we remove an osrfRouterClass from the
 631         router's list of classes.
 632 */
 633 static void osrfRouterClassFree( char* classname, void* c ) {
 634         if( !c )
 635                 return;
 636         osrfRouterClass* rclass = (osrfRouterClass*) c;
 637         client_disconnect( rclass->connection );
 638         client_free( rclass->connection );
 639
 640         osrfHashIteratorReset( rclass->itr );
 641         osrfRouterNode* node;
 642
 643         while( (node = osrfHashIteratorNext(rclass->itr)) )
 644                 osrfHashRemove( rclass->nodes, node->remoteId );
 645
 646         osrfHashIteratorFree(rclass->itr);
 647         osrfHashFree(rclass->nodes);
 648
 649         free(rclass);
 650 }
 651
 652 /**
 653         @brief Free an osrfRouterNode.
 654         @param remoteId Jabber ID of node (not used).
 655         @param n Pointer to the osrfRouterNode to be freed, cast to a void pointer.
 656
 657         This is a callback installed in an osrfHash (the nodes member of an osrfRouterClass).
 658 */
 659 static void osrfRouterNodeFree( char* remoteId, void* n ) {
 660         if(!n) return;
 661         osrfRouterNode* node = (osrfRouterNode*) n;
 662         free(node->remoteId);
 663         message_free(node->lastMessage);
 664         free(node);
 665 }
 666
 667
 668 /**
 669         @brief Free an osrfRouter and everything it owns.
 670         @param router Pointer to the osrfRouter to be freed.
 671
 672         The osrfRouterClasses and osrfRouterNodes are freed by callback functions installed in
 673         the osrfHashes.
 674 */
 675 void osrfRouterFree( osrfRouter* router ) {
 676         if(!router) return;
 677
 678         osrfHashIteratorFree( router->class_itr);
 679         osrfHashFree(router->classes);
 680         free(router->domain);
 681         free(router->name);
 682         free(router->resource);
 683         free(router->password);
 684
 685         osrfStringArrayFree( router->trustedClients );
 686         osrfStringArrayFree( router->trustedServers );
 687
 688         client_free( router->connection );
 689         free(router);
 690 }
 691
 692
 693 /**
 694         @brief Given a class name, find the corresponding osrfRouterClass.
 695         @param router Pointer to the osrfRouter that owns the osrfRouterClass.
 696         @param classname Name of the class.
 697         @return Pointer to a matching osrfRouterClass if found, or NULL if not.
 698 */
 699 static osrfRouterClass* osrfRouterFindClass( osrfRouter* router, const char* classname ) {
 700         if(!( router && router->classes && classname )) return NULL;
 701         return (osrfRouterClass*) osrfHashGet( router->classes, classname );
 702 }
 703
 704
 705 /**
 706         @brief Find a given node for a given class.
 707         @param rclass Pointer to the osrfRouterClass in which to search.
 708         @param remoteId Jabber ID of the node for which to search.
 709         @return Pointer to the matching osrfRouterNode, if found; otherwise NULL.
 710 */
 711 static osrfRouterNode* osrfRouterClassFindNode( osrfRouterClass* rclass,
 712                 const char* remoteId ) {
 713         if(!(rclass && remoteId))  return NULL;
 714         return (osrfRouterNode*) osrfHashGet( rclass->nodes, remoteId );
 715 }
 716
 717
 718 /**
 719         @brief Fill an fd_set with all the sockets owned by the osrfRouter.
 720         @param router Pointer to the osrfRouter whose sockets are to be used.
 721         @param set Pointer to the fd_set that is to be filled.
 722         @return The largest file descriptor loaded into the fd_set; or -1 upon error.
 723
 724         There's one socket for the osrfRouter as a whole, and one for each osrfRouterClass
 725         that belongs to it.  We load them all.
 726 */
 727 static int _osrfRouterFillFDSet( osrfRouter* router, fd_set* set ) {
 728         if(!(router && router->classes && set)) return -1;
 729
 730         FD_ZERO(set);
 731         int maxfd = client_sock_fd( router->connection );
 732         FD_SET(maxfd, set);
 733
 734         int sockid;
 735
 736         osrfRouterClass* class = NULL;
 737         osrfHashIterator* itr = router->class_itr;
 738         osrfHashIteratorReset( itr );
 739
 740         while( (class = osrfHashIteratorNext(itr)) ) {
 741                 const char* classname = osrfHashIteratorKey(itr);
 742
 743                 if( classname && (class = osrfRouterFindClass( router, classname )) ) {
 744                         sockid = client_sock_fd( class->connection );
 745
 746                         if( osrfUtilsCheckFileDescriptor( sockid ) ) {
 747
 748                                 osrfLogWarning(OSRF_LOG_MARK,
 749                                         "Removing router class '%s' because of a bad top-level file descriptor [%d]",
 750                                         classname, sockid );
 751                                 osrfRouterRemoveClass( router, classname );
 752
 753                         } else {
 754                                 if( sockid > maxfd ) maxfd = sockid;
 755                                 FD_SET(sockid, set);
 756                         }
 757                 }
 758         }
 759
 760         return maxfd;
 761 }
 762
 763 /**
 764         @brief Handler a router-level message that isn't a command; presumed to be an app request.
 765         @param router Pointer to the current osrfRouter.
 766         @param msg Pointer to the incoming message.
 767
 768         The body of the transport_message is a JSON string, specifically an JSON array.
 769         Translate the JSON into a series of osrfMessages, one for each element of the JSON
 770         array.  Process each osrfMessage in turn.  Each message is either a CONNECT or a
 771         REQUEST.
 772 */
 773 static void osrfRouterHandleAppRequest( osrfRouter* router, const transport_message* msg ) {
 774
 775         int T = 32;
 776         osrfMessage* arr[T];
 777
 778         // Initialize pointer array to all NULLs
 779         int i;
 780         for( i = 0; i < T; ++i )
 781                 arr[ i ] = NULL;
 782
 783         // Translate the JSON into an array of pointers to osrfMessage
 784         int num_msgs = osrf_message_deserialize( msg->body, arr, T );
 785         osrfMessage* omsg = NULL;
 786
 787         // Process each osrfMessage
 788         for( i = 0; i < num_msgs; i++ ) {
 789
 790                 omsg = arr[i];
 791                 if( omsg ) {
 792
 793                         switch( omsg->m_type ) {
 794
 795                                 case CONNECT:
 796                                         osrfRouterRespondConnect( router, msg, omsg );
 797                                         break;
 798
 799                                 case REQUEST:
 800                                         osrfRouterProcessAppRequest( router, msg, omsg );
 801                                         break;
 802
 803                                 default:
 804                                         break;
 805                         }
 806
 807                         osrfMessageFree( omsg );
 808                 }
 809         }
 810
 811         return;
 812 }
 813
 814 /**
 815         @brief Respond to a CONNECT message.
 816         @param router Pointer to the current osrfRouter.
 817         @param msg Pointer to the transport_message that the osrfMessage came from.
 818         @param omsg Pointer to the osrfMessage to be processed.
 819
 820         An application is trying to connect to the router.  Reply with a STATUS message
 821         signifying success.
 822
 823         "CONNECT" is a bit of a misnomer.  We don't establish a stateful session; i.e. we
 824         don't retain any memory of this message.  We just confirm that the router is alive,
 825         and that the client has a good address for it.
 826 */
 827 static void osrfRouterRespondConnect( osrfRouter* router, const transport_message* msg,
 828                 const osrfMessage* omsg ) {
 829         if(!(router && msg && omsg))
 830                 return;
 831
 832         osrfLogDebug( OSRF_LOG_MARK, "router received a CONNECT message from %s", msg->sender );
 833
 834         // Build a success message
 835         osrfMessage* success = osrf_message_init( STATUS, omsg->thread_trace, omsg->protocol );
 836         osrf_message_set_status_info(
 837                 success, "osrfConnectStatus", "Connection Successful", OSRF_STATUS_OK );
 838
 839         // Translate the success message into JSON,
 840         // and then package the JSON in a transport message
 841         char* data = osrf_message_serialize(success);
 842         osrfMessageFree( success );
 843         transport_message* return_msg = message_init(
 844                 data,           // message payload, translated from osrfMessage
 845                 "",             // no subject
 846                 msg->thread,    // same thread
 847                 msg->sender,    // destination (client's Jabber ID)
 848                 ""              // don't send our address; client already has it
 849         );
 850         free( data );
 851
 852         client_send_message( router->connection, return_msg );
 853         message_free( return_msg );
 854 }
 855
 856
 857 /**
 858         @brief Respond to a REQUEST message.
 859         @param router Pointer to the current osrfRouter.
 860         @param msg Pointer to the transport_message that the osrfMessage came from.
 861         @param omsg Pointer to the osrfMessage to be processed.
 862
 863         Respond to an information request from an application.  Most types of request involve
 864         one or more counts of messages successfully routed.
 865
 866         Request types currently supported:
 867         - "opensrf.router.info.class.list" -- list of class names.
 868         - "opensrf.router.info.stats.class.summary" -- total count for a specified class.
 869         - "opensrf.router.info.stats.class" -- count for every node of a specified class.
 870         - "opensrf.router.info.stats.class.all" -- count for every node of every class.
 871         - "opensrf.router.info.stats.class.node.all" -- total count for every class.
 872 */
 873 static void osrfRouterProcessAppRequest( osrfRouter* router, const transport_message* msg,
 874                 const osrfMessage* omsg ) {
 875
 876         if(!(router && msg && omsg && omsg->method_name))
 877                 return;
 878
 879         osrfLogInfo( OSRF_LOG_MARK, "Router received app request: %s", omsg->method_name );
 880
 881         // Branch on the request type.  Build a jsonObject as an answer to the request.
 882         jsonObject* jresponse = NULL;
 883         if(!strcmp( omsg->method_name, ROUTER_REQUEST_CLASS_LIST )) {
 884
 885                 // Prepare an array of class names.
 886                 int i;
 887                 jresponse = jsonNewObjectType(JSON_ARRAY);
 888
 889                 osrfStringArray* keys = osrfHashKeys( router->classes );
 890                 for( i = 0; i != keys->size; i++ )
 891                         jsonObjectPush( jresponse, jsonNewObject(osrfStringArrayGetString( keys, i )) );
 892                 osrfStringArrayFree(keys);
 893
 894         } else if(!strcmp( omsg->method_name, ROUTER_REQUEST_STATS_CLASS_SUMMARY )) {
 895
 896                 // Prepare a count of all the messages successfully routed for a given class.
 897                 int count = 0;
 898
 899                 // class name is the first parameter
 900                 const char* classname = jsonObjectGetString( jsonObjectGetIndex( omsg->_params, 0 ) );
 901                 if (!classname)
 902                         return;
 903
 904                 osrfRouterClass* class = osrfHashGet(router->classes, classname);
 905
 906                 // For each node: add the count to the total.
 907                 osrfRouterNode* node;
 908                 osrfHashIterator* node_itr = osrfNewHashIterator(class->nodes);
 909                 while( (node = osrfHashIteratorNext(node_itr)) ) {
 910                         count += node->count;
 911                         // jsonObjectSetKey( class_res, node->remoteId,
 912                         //       jsonNewNumberObject( (double) node->count ) );
 913                 }
 914                 osrfHashIteratorFree(node_itr);
 915
 916                 jresponse = jsonNewNumberObject( (double) count );
 917
 918         } else if(!strcmp( omsg->method_name, ROUTER_REQUEST_STATS_CLASS )) {
 919
 920                 // Prepare a hash for a given class.  Key: the remoteId of a node.  Datum: the
 921                 // number of messages successfully routed for that node.
 922
 923                 // class name is the first parameter
 924                 const char* classname = jsonObjectGetString( jsonObjectGetIndex( omsg->_params, 0 ) );
 925                 if (!classname)
 926                         return;
 927
 928                 jresponse = jsonNewObjectType(JSON_HASH);
 929                 osrfRouterClass* class = osrfHashGet(router->classes, classname);
 930
 931                 // For each node: get the count and store it in the hash.
 932                 osrfRouterNode* node;
 933                 osrfHashIterator* node_itr = osrfNewHashIterator(class->nodes);
 934                 while( (node = osrfHashIteratorNext(node_itr)) ) {
 935                         jsonObjectSetKey( jresponse, node->remoteId,
 936                                         jsonNewNumberObject( (double) node->count ) );
 937                 }
 938                 osrfHashIteratorFree(node_itr);
 939
 940         } else if(!strcmp( omsg->method_name, ROUTER_REQUEST_STATS_CLASS_FULL )) {
 941
 942                 // Prepare a hash of hashes, giving the message counts for each node for each class.
 943
 944                 osrfRouterClass* class;
 945                 osrfRouterNode* node;
 946                 jresponse = jsonNewObjectType(JSON_HASH);  // Key: class name.
 947
 948                 // Traverse the list of classes.
 949                 osrfHashIterator* class_itr = osrfNewHashIterator(router->classes);
 950                 while( (class = osrfHashIteratorNext(class_itr)) ) {
 951
 952                         jsonObject* class_res = jsonNewObjectType(JSON_HASH);  // Key: remoteId of node.
 953                         const char* classname = osrfHashIteratorKey(class_itr);
 954
 955                         // Traverse the list of nodes for the current class.
 956                         osrfHashIterator* node_itr = osrfNewHashIterator(class->nodes);
 957                         while( (node = osrfHashIteratorNext(node_itr)) ) {
 958                                 jsonObjectSetKey( class_res, node->remoteId,
 959                                                 jsonNewNumberObject( (double) node->count ) );
 960                         }
 961                         osrfHashIteratorFree(node_itr);
 962
 963                         jsonObjectSetKey( jresponse, classname, class_res );
 964                 }
 965
 966                 osrfHashIteratorFree(class_itr);
 967
 968         } else if(!strcmp( omsg->method_name, ROUTER_REQUEST_STATS_NODE_FULL )) {
 969
 970                 // Prepare a hash. Key: class name.  Datum: total number of successfully routed
 971                 // messages routed for nodes of that class.
 972
 973                 osrfRouterClass* class;
 974                 osrfRouterNode* node;
 975                 jresponse = jsonNewObjectType(JSON_HASH);
 976
 977                 osrfHashIterator* class_itr = osrfNewHashIterator(router->classes);
 978                 while( (class = osrfHashIteratorNext(class_itr)) ) {  // For each class
 979
 980                         int count = 0;
 981                         const char* classname = osrfHashIteratorKey(class_itr);
 982
 983                         osrfHashIterator* node_itr = osrfNewHashIterator(class->nodes);
 984                         while( (node = osrfHashIteratorNext(node_itr)) ) {  // For each node
 985                                 count += node->count;
 986                         }
 987                         osrfHashIteratorFree(node_itr);
 988
 989                         jsonObjectSetKey( jresponse, classname, jsonNewNumberObject( (double) count ) );
 990                 }
 991
 992                 osrfHashIteratorFree(class_itr);
 993
 994         } else {  // None of the above
 995
 996                 osrfRouterHandleMethodNFound( router, msg, omsg );
 997                 return;
 998         }
 999
1000         // Send the result back to the requester.
1001         osrfRouterSendAppResponse( router, msg, omsg, jresponse );
1002         jsonObjectFree(jresponse);
1003 }
1004
1005
1006 /**
1007         @brief Respond to an invalid REQUEST message.
1008         @param router Pointer to the current osrfRouter.
1009         @param msg Pointer to the transport_message that contained the REQUEST message.
1010         @param omsg Pointer to the osrfMessage that contained the REQUEST.
1011 */
1012 static void osrfRouterHandleMethodNFound( osrfRouter* router,
1013                 const transport_message* msg, const osrfMessage* omsg ) {
1014
1015         // Create an exception message
1016         osrfMessage* err = osrf_message_init( STATUS, omsg->thread_trace, 1 );
1017         osrf_message_set_status_info( err,
1018                         "osrfMethodException", "Router method not found", OSRF_STATUS_NOTFOUND );
1019
1020         // Translate it into JSON
1021         char* data =  osrf_message_serialize(err);
1022         osrfMessageFree( err );
1023
1024         // Wrap the JSON up in a transport_message
1025         transport_message* tresponse = message_init(
1026                         data, "", msg->thread, msg->sender, msg->recipient );
1027         free(data);
1028
1029         // Send it
1030         client_send_message( router->connection, tresponse );
1031         message_free( tresponse );
1032 }
1033
1034
1035 /**
1036         @brief Send a response to a router REQUEST message.
1037         @param router Pointer to the current osrfRouter.
1038         @param msg Pointer to the transport_message that contained the REQUEST message.
1039         @param omsg Pointer to the osrfMessage that contained the REQUEST.
1040         @param response Pointer to the jsonObject that constitutes the response.
1041 */
1042 static void osrfRouterSendAppResponse( osrfRouter* router, const transport_message* msg,
1043                 const osrfMessage* omsg, const jsonObject* response ) {
1044
1045         if( response ) { /* send the response message */
1046
1047                 // Turn the jsonObject into JSON, and load it into an osrfMessage
1048                 osrfMessage* oresponse = osrf_message_init(
1049                                 RESULT, omsg->thread_trace, omsg->protocol );
1050
1051                 char* json = jsonObjectToJSON(response);
1052                 osrf_message_set_result_content( oresponse, json );
1053                 free(json);
1054
1055                 // Package the osrfMessage into a transport_message, and send it
1056                 char* data =  osrf_message_serialize(oresponse);
1057                 osrfMessageFree(oresponse);
1058                 osrfLogDebug( OSRF_LOG_MARK,  "Responding to client app request with data: \n%s\n", data );
1059
1060                 transport_message* tresponse = message_init(
1061                                 data, "", msg->thread, msg->sender, msg->recipient );
1062                 free(data);
1063
1064                 client_send_message(router->connection, tresponse );
1065                 message_free(tresponse);
1066         }
1067
1068         /* now send the 'request complete' message */
1069         osrfMessage* status = osrf_message_init( STATUS, omsg->thread_trace, 1);
1070         osrf_message_set_status_info( status, "osrfConnectStatus", "Request Complete",
1071                         OSRF_STATUS_COMPLETE );
1072         char* statusdata = osrf_message_serialize(status);
1073         osrfMessageFree(status);
1074
1075         transport_message* sresponse = message_init(
1076                         statusdata, "", msg->thread, msg->sender, msg->recipient );
1077         free(statusdata);
1078
1079         client_send_message(router->connection, sresponse );
1080         message_free(sresponse);
1081 }