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->nodes                                      = osrfNewHash();
  29         grp->itr                                                = osrfNewHashIterator(grp->nodes);
  30
  31         int i;
  32         for( i = 0; i != count; i++ ) {
  33                 if(!(nodes[i] && nodes[i]->domain) ) return NULL;
  34                 osrfHashSet( grp->nodes, nodes[i], nodes[i]->domain );
  35         }
  36
  37         return grp;
  38 }
  39
  40
  41 /* connect all of the nodes to their servers */
  42 int osrfTransportGroupConnect( osrfTransportGroup* grp ) {
  43         if(!grp) return -1;
  44         int active = 0;
  45
  46         osrfTransportGroupNode* node;
  47         osrfHashIteratorReset(grp->itr);
  48
  49         while( (node = osrfHashIteratorNext(grp->itr)) ) {
  50                 if(client_connect( node->connection, node->username,
  51                                         node->password, node->resource, 10, AUTH_DIGEST )) {
  52                         node->active = 1;
  53                         active++;
  54                 }
  55         }
  56
  57         osrfHashIteratorReset(grp->itr);
  58         return active;
  59 }
  60
  61
  62 int osrfTransportGroupSendMatch( osrfTransportGroup* grp, transport_message* msg ) {
  63         if(!(grp && msg)) return -1;
  64
  65         char domain[256];
  66         bzero(domain, 256);
  67         jid_get_domain( msg->recipient, domain, 255 );
  68
  69         osrfTransportGroupNode* node = osrfHashGet(grp->nodes, domain);
  70         if(node) {
  71                 if( (client_send_message( node->connection, msg )) == 0 )
  72                         return 0;
  73         }
  74
  75         return warning_handler("Error sending message to domain %s", domain );
  76 }
  77
  78 int osrfTransportGroupSend( osrfTransportGroup* grp, transport_message* msg ) {
  79
  80         if(!(grp && msg)) return -1;
  81         int bufsize = 256;
  82
  83         char domain[bufsize];
  84         bzero(domain, bufsize);
  85         jid_get_domain( msg->recipient, domain, bufsize - 1 );
  86
  87         char msgrecip[bufsize];
  88         bzero(msgrecip, bufsize);
  89         jid_get_username(msg->recipient, msgrecip, bufsize - 1);
  90
  91         char msgres[bufsize];
  92         bzero(msgres, bufsize);
  93         jid_get_resource(msg->recipient, msgres, bufsize - 1);
  94
  95         char* firstdomain = NULL;
  96         char newrcp[1024];
  97
  98
  99         int updateRecip = 1;
 100         /* if we don't host this domain, don't update the recipient but send it as is */
 101         if(osrfHashGet(grp->nodes, domain)) updateRecip = 0;
 102
 103         osrfTransportGroupNode* node;
 104
 105         do {
 106
 107                 node = osrfHashIteratorNext(grp->itr);
 108                 if(!node) osrfHashIteratorReset(grp->itr);
 109
 110                 node = osrfHashIteratorNext(grp->itr);
 111                 if(!node) return -1;
 112
 113                 if(firstdomain == NULL) {
 114                         firstdomain = node->domain;
 115
 116                 } else {
 117                         if(!strcmp(firstdomain, node->domain)) { /* we've made a full loop */
 118                                 return warning_handler("We've tried to send to all domains.. giving up");
 119                         }
 120                 }
 121
 122                 /* update the recipient domain if necessary */
 123                 bzero(newrcp, 1024);
 124                 if(updateRecip)
 125                         sprintf(newrcp, "%s@%s/%s", msgrecip, node->domain, msgres);
 126                 else
 127                         sprintf(newrcp, msg->recipient);
 128
 129                 free(msg->recipient);
 130                 msg->recipient = strdup(newrcp);
 131
 132                 if( (client_send_message( node->connection, msg )) == 0 )
 133                         return 0;
 134
 135         } while(1);
 136
 137         return -1;
 138 }
 139
 140 static int __osrfTGWait( fd_set* fdset, int maxfd, int timeout ) {
 141         if(!(fdset && maxfd)) return 0;
 142
 143         struct timeval tv;
 144         tv.tv_sec = timeout;
 145         tv.tv_usec = 0;
 146         int retval = 0;
 147
 148         if( timeout < 0 ) {
 149                 if( (retval = select( maxfd + 1, fdset, NULL, NULL, NULL)) == -1 )
 150                         return 0;
 151
 152         } else {
 153                 if( (retval = select( maxfd + 1, fdset, NULL, NULL, &tv)) == -1 )
 154                         return 0;
 155         }
 156
 157         return retval;
 158 }
 159
 160
 161 transport_message* osrfTransportGroupRecvAll( osrfTransportGroup* grp, int timeout ) {
 162         if(!grp) return NULL;
 163
 164         int maxfd = 0;
 165         fd_set fdset;
 166         FD_ZERO( &fdset );
 167
 168         osrfTransportGroupNode* node;
 169         osrfHashIterator* itr = osrfNewHashIterator(grp->nodes);
 170
 171         while( (node = osrfHashIteratorNext(itr)) ) {
 172                 if(node->active) {
 173                         int fd = node->connection->session->sock_id;
 174                         if( fd < maxfd ) maxfd = fd;
 175                         FD_SET( fd, &fdset );
 176                 }
 177         }
 178         osrfHashIteratorReset(itr);
 179
 180         if( __osrfTGWait( &fdset, maxfd, timeout ) ) {
 181                 while( (node = osrfHashIteratorNext(itr)) ) {
 182                         if(node->active) {
 183                                 int fd = node->connection->session->sock_id;
 184                                 if( FD_ISSET( fd, &fdset ) ) {
 185                                         return client_recv( node->connection, 0 );
 186                                 }
 187                         }
 188                 }
 189         }
 190
 191         osrfHashIteratorFree(itr);
 192         return NULL;
 193 }
 194
 195 transport_message* osrfTransportGroupRecv( osrfTransportGroup* grp, char* domain, int timeout ) {
 196         if(!(grp && domain)) return NULL;
 197
 198         osrfTransportGroupNode* node = osrfHashGet(grp->nodes, domain);
 199         if(!node && node->connection && node->connection->session) return NULL;
 200         int fd = node->connection->session->sock_id;
 201
 202         fd_set fdset;
 203         FD_ZERO( &fdset );
 204         FD_SET( fd, &fdset );
 205
 206         int active = __osrfTGWait( &fdset, fd, timeout );
 207         if(active) return client_recv( node->connection, 0 );
 208
 209         return NULL;
 210 }
 211
 212 void osrfTransportGroupSetInactive( osrfTransportGroup* grp, char* domain ) {
 213         if(!(grp && domain)) return;
 214         osrfTransportGroupNode* node = osrfHashGet(grp->nodes, domain );
 215         if(node) node->active = 0;
 216 }
 217
 218 /*
 219 osrfTransportGroupNode* __osrfTransportGroupFindNode( osrfTransportGroup* grp, char* domain ) {
 220         if(!(grp && grp->list && domain)) return NULL;
 221         int i = 0;
 222         osrfTransportGroupNode* node = NULL;
 223
 224         while( (node = (osrfTransportGroupNode*) osrfListGetIndex( grp->list, i++ )) )
 225                 if(!strcmp(node->domain, domain)) return node;
 226         return NULL;
 227 }
 228 */
 229
 230
 231
 232