OpenSRF/src/libstack/osrf_transgroup.c

   1 #include "osrf_transgroup.h"
   2 #include <sys/select.h>
   3
   4
   5 osrfTransportGroupNode* osrfNewTransportGroupNode(
   6                 char* domain, int port, char* username, char* password, char* resource ) {
   7
   8         if(!(domain && port && username && password && resource)) return NULL;
   9
  10         osrfTransportGroupNode* node = safe_malloc(sizeof(osrfTransportGroupNode));
  11         node->domain    = strdup(domain);
  12         node->port              = port;
  13         node->username = strdup(username);
  14         node->password = strdup(password);
  15         node->domain    = strdup(domain);
  16         node->active    = 0;
  17         node->lastsent  = 0;
  18         node->connection = client_init( domain, port, NULL, 0 );
  19
  20         return node;
  21 }
  22
  23
  24 osrfTransportGroup* osrfNewTransportGroup( char* router, osrfTransportGroupNode* nodes[], int count ) {
  25         if(!nodes || !router || count < 1) return NULL;
  26
  27         osrfTransportGroup* grp = safe_malloc(sizeof(osrfTransportGroup));
  28         grp->currentNode                        = 0;
  29         grp->router                                     = strdup(router);
  30         grp->list                                       = osrfNewList(1);
  31
  32         int i;
  33         for( i = 0; i != count; i++ ) osrfListPush( grp->list, nodes[i] );
  34         return grp;
  35 }
  36
  37
  38 int osrfTransportGroupConnect( osrfTransportGroup* grp ) {
  39         if(!grp) return 0;
  40         int i;
  41         int active = 0;
  42         for( i = 0; i != grp->list->size; i++ ) {
  43                 osrfTransportGroupNode* node = osrfListGetIndex( grp->list, i );
  44                 if(client_connect( node->connection, node->username,
  45                                         node->password, node->resource, 10, AUTH_DIGEST )) {
  46                         node->active = 1;
  47                         node->lastsent = time(NULL);
  48                         active++;
  49                 }
  50         }
  51         return active;
  52 }
  53
  54
  55 /*
  56 osrfTransportGroup* osrfNewTransportGroup( char* resource ) {
  57
  58         grp->username                           = osrfConfigGetValue( NULL, "/username" );
  59         grp->password                           = osrfConfigGetValue( NULL, "/passwd" );
  60         char* port                                      = osrfConfigGetValue( NULL, "/port" );
  61         if(port) grp->port              = atoi(port);
  62         grp->currentNode                        = 0;
  63
  64         if(!resource) resource = "client";
  65         char* host = getenv("HOSTNAME");
  66         if(!host) host = "localhost";
  67         char* res = va_list_to_string( "osrf_%s_%s_%d", resource, host, getpid() );
  68
  69         int i;
  70         osrfStringArray* arr = osrfNewStringArray(8);
  71         osrfConfigGetValueList(NULL, arr, "/domains/domain");
  72
  73         for( i = 0; i != arr->size; i++ ) {
  74                 char* domain = osrfStringArrayGetString( arr, i );
  75                 if(domain) {
  76                         node->domain = strdup(domain);
  77                         node->connection = client_init( domain, grp->port, NULL, 0 );
  78                         if(client_connect( node->connection, grp->username, grp->password, res, 10, AUTH_DIGEST )) {
  79                                 node->active = 1;
  80                                 node->lastsent = time(NULL);
  81                         }
  82                         osrfListPush( grp->list, node );
  83                 }
  84         }
  85
  86         free(res);
  87         osrfStringArrayFree(arr);
  88         return grp;
  89 }
  90 */
  91
  92
  93 int osrfTransportGroupSend( osrfTransportGroup* grp, transport_message* msg, char* newdomain ) {
  94         if(!(grp && msg)) return -1;
  95
  96         char domain[256];
  97         bzero(domain, 256);
  98         jid_get_domain( msg->recipient, domain );
  99
 100         char msgrecip[254];
 101         bzero(msgrecip, 254);
 102         jid_get_username(msg->recipient, msgrecip);
 103
 104
 105         osrfTransportGroupNode* node = __osrfTransportGroupFindNode( grp, domain );
 106
 107         if( strcmp( msgrecip, grp->router ) ) { /* not a top level router message */
 108
 109                 if(node) {
 110                         if( (client_send_message( node->connection, msg )) == 0 )
 111                                 return 0;
 112                         else
 113                                 return warning_handler("Error sending message to domain %s", domain );
 114                 }
 115                 return warning_handler("Transport group has no node for domain %s", domain );
 116         }
 117
 118
 119         /*
 120         if( type == OSRF_SERVER_NODE )
 121                 return _osrfTGServerSend( grp, msgdom, msg );
 122         if( type == OSRF_CLIENT_NODE )
 123                 return _osrfTGClientSend( grp, msgdom, msg );
 124                 */
 125
 126         return -1;
 127 }
 128
 129 int _osrfTGServerSend( osrfTransportGroup* grp, char* domain, transport_message* msg ) {
 130
 131         debug_handler("Transport group sending server message to domain %s", domain );
 132
 133         osrfTransportGroupNode* node = __osrfTransportGroupFindNode( grp, domain );
 134         if(node) {
 135                 if( (client_send_message( node->connection, msg )) == 0 )
 136                         return 0;
 137                 else
 138                         return warning_handler("Error sending server response to domain %s", domain );
 139         }
 140         return warning_handler("Transport group has no node for domain %s for server response", domain );
 141 }
 142
 143
 144 int _osrfTGClientSend( osrfTransportGroup* grp, char* domain, transport_message* msg ) {
 145
 146         debug_handler("Transport group sending client message to domain %s", domain );
 147
 148         /* first see if we have a node for the requested domain */
 149         osrfTransportGroupNode* node = __osrfTransportGroupFindNode( grp, domain );
 150         if(node && node->active) {
 151                 if( (client_send_message( node->connection, msg )) == 0 )
 152                         return 0;
 153                 else
 154                         node->active = 0;
 155         }
 156
 157         /* if not (or it fails), try sending to the current domain */
 158         node = osrfListGetIndex(grp->list, grp->currentNode);
 159         if(node && node->active) {
 160                 if( (client_send_message( node->connection, msg )) == 0 )
 161                         return 0;
 162         }
 163
 164         /* start at the beginning and try them all ... */
 165         grp->currentNode = 0;
 166         while( grp->currentNode < grp->list->size ) {
 167                 if( (node = osrfListGetIndex(grp->list, grp->currentNode++)) && node->active ) {
 168                         if( (client_send_message( node->connection, msg )) == 0 )
 169                                 return 1;
 170                         else node->active = 0;
 171                 }
 172         }
 173         return -1;
 174 }
 175
 176 static int __osrfTGWait( fd_set* fdset, int maxfd, int timeout ) {
 177         if(!(fdset && maxfd)) return 0;
 178
 179         struct timeval tv;
 180         tv.tv_sec = timeout;
 181         tv.tv_usec = 0;
 182         int retval = 0;
 183
 184         if( timeout < 0 ) {
 185                 if( (retval = select( maxfd + 1, fdset, NULL, NULL, NULL)) == -1 )
 186                         return 0;
 187
 188         } else {
 189                 if( (retval = select( maxfd + 1, fdset, NULL, NULL, &tv)) == -1 )
 190                         return 0;
 191         }
 192
 193         return retval;
 194 }
 195
 196
 197 transport_message* osrfTransportGroupRecvAll( osrfTransportGroup* grp, int timeout ) {
 198         if(!(grp && grp->list)) return NULL;
 199
 200         int i;
 201         int maxfd = 0;
 202         osrfTransportGroupNode* node = NULL;
 203         fd_set fdset;
 204         FD_ZERO( &fdset );
 205
 206         for( i = 0; i != grp->list->size; i++ ) {
 207                 if( (node = osrfListGetIndex(grp->list, grp->currentNode++)) && node->active ) {
 208                         int fd = node->connection->session->sock_id;
 209                         if( fd < maxfd ) maxfd = fd;
 210                         FD_SET( fd, &fdset );
 211                 }
 212         }
 213
 214         if( __osrfTGWait( &fdset, maxfd, timeout ) ) {
 215                 for( i = 0; i != grp->list->size; i++ ) {
 216                         if( (node = osrfListGetIndex(grp->list, grp->currentNode++)) && node->active ) {
 217                                 int fd = node->connection->session->sock_id;
 218                                 if( FD_ISSET( fd, &fdset ) ) {
 219                                         return client_recv( node->connection, 0 );
 220                                 }
 221                         }
 222                 }
 223         }
 224
 225         return NULL;
 226 }
 227
 228 transport_message* osrfTransportGroupRecv( osrfTransportGroup* grp, char* domain, int timeout ) {
 229         if(!(grp && domain)) return NULL;
 230
 231         osrfTransportGroupNode* node = __osrfTransportGroupFindNode( grp, domain );
 232         if(!node && node->connection && node->connection->session) return NULL;
 233         int fd = node->connection->session->sock_id;
 234
 235         fd_set fdset;
 236         FD_ZERO( &fdset );
 237         FD_SET( fd, &fdset );
 238
 239         int active = __osrfTGWait( &fdset, fd, timeout );
 240         if(active) return client_recv( node->connection, 0 );
 241
 242         return NULL;
 243 }
 244
 245 void osrfTransportGroupSetInactive( osrfTransportGroup* grp, char* domain ) {
 246         if(!(grp && domain)) return;
 247         osrfTransportGroupNode* node = __osrfTransportGroupFindNode( grp, domain );
 248         if(node) node->active = 0;
 249 }
 250
 251 osrfTransportGroupNode* __osrfTransportGroupFindNode( osrfTransportGroup* grp, char* domain ) {
 252         if(!(grp && grp->list && domain)) return NULL;
 253         int i = 0;
 254         osrfTransportGroupNode* node = NULL;
 255
 256         while( (node = (osrfTransportGroupNode*) osrfListGetIndex( grp->list, i++ )) )
 257                 if(!strcmp(node->domain, domain)) return node;
 258         return NULL;
 259 }
 260
 261
 262
 263