src/libopensrf/transport_client.c

   1 #include <opensrf/transport_client.h>
   2
   3
   4 //int main( int argc, char** argv );
   5
   6 /*
   7 int main( int argc, char** argv ) {
   8
   9         transport_message* recv;
  10         transport_message* send;
  11
  12         transport_client* client = client_init( "spacely.georgialibraries.org", 5222 );
  13
  14         // try to connect, allow 15 second connect timeout
  15         if( client_connect( client, "admin", "asdfjkjk", "system", 15 ) ) {
  16                 printf("Connected...\n");
  17         } else {
  18                 printf( "NOT Connected...\n" ); exit(99);
  19         }
  20
  21         while( (recv = client_recv( client, -1 )) ) {
  22
  23                 if( recv->body ) {
  24                         int len = strlen(recv->body);
  25                         char buf[len + 20];
  26                         osrf_clearbuf( buf, 0, sizeof(buf));
  27                         sprintf( buf, "Echoing...%s", recv->body );
  28                         send = message_init( buf, "Echoing Stuff", "12345", recv->sender, "" );
  29                 } else {
  30                         send = message_init( " * ECHOING * ", "Echoing Stuff", "12345", recv->sender, "" );
  31                 }
  32
  33                 if( send == NULL ) { printf("something's wrong"); }
  34                 client_send_message( client, send );
  35
  36                 message_free( send );
  37                 message_free( recv );
  38         }
  39
  40         printf( "ended recv loop\n" );
  41
  42         return 0;
  43
  44 }
  45 */
  46
  47
  48 transport_client* client_init( const char* server, int port, const char* unix_path, int component ) {
  49
  50         if(server == NULL) return NULL;
  51
  52         /* build and clear the client object */
  53         transport_client* client = safe_malloc( sizeof( transport_client) );
  54
  55         /* build and clear the message list */
  56         client->m_list = safe_malloc( sizeof( transport_message_list ) );
  57
  58         client->m_list->next = NULL;
  59         client->m_list->message = NULL;
  60         client->m_list->type = MESSAGE_LIST_HEAD;
  61
  62         /* build the session */
  63
  64         client->session = init_transport( server, port, unix_path, client, component );
  65
  66         client->session->message_callback = client_message_handler;
  67         client->error = 0;
  68
  69         return client;
  70 }
  71
  72
  73 int client_connect( transport_client* client,
  74                 const char* username, const char* password, const char* resource,
  75                 int connect_timeout, enum TRANSPORT_AUTH_TYPE  auth_type ) {
  76         if(client == NULL) return 0;
  77         return session_connect( client->session, username,
  78                         password, resource, connect_timeout, auth_type );
  79 }
  80
  81
  82 int client_disconnect( transport_client* client ) {
  83         if( client == NULL ) { return 0; }
  84         return session_disconnect( client->session );
  85 }
  86
  87 int client_connected( const transport_client* client ) {
  88         if(client == NULL) return 0;
  89         return client->session->state_machine->connected;
  90 }
  91
  92 int client_send_message( transport_client* client, transport_message* msg ) {
  93         if(client == NULL) return 0;
  94         if( client->error ) return -1;
  95         return session_send_msg( client->session, msg );
  96 }
  97
  98
  99 transport_message* client_recv( transport_client* client, int timeout ) {
 100         if( client == NULL ) { return NULL; }
 101
 102         transport_message_node* node;
 103         transport_message* msg;
 104
 105
 106         /* see if there are any message in the messages queue */
 107         if( client->m_list->next != NULL ) {
 108                 /* pop off the first one... */
 109                 node = client->m_list->next;
 110                 client->m_list->next = node->next;
 111                 msg = node->message;
 112                 free( node );
 113                 return msg;
 114         }
 115
 116         if( timeout == -1 ) {  /* wait potentially forever for data to arrive */
 117
 118                 while( client->m_list->next == NULL ) {
 119                 //      if( ! session_wait( client->session, -1 ) ) {
 120                         int x;
 121                         if( (x = session_wait( client->session, -1 )) ) {
 122                                 osrfLogDebug(OSRF_LOG_MARK, "session_wait returned failure code %d\n", x);
 123                                 client->error = 1;
 124                                 return NULL;
 125                         }
 126                 }
 127
 128         } else { /* wait at most timeout seconds */
 129
 130
 131                 /* if not, loop up to 'timeout' seconds waiting for data to arrive */
 132                 time_t start = time(NULL);
 133                 time_t remaining = (time_t) timeout;
 134
 135                 int counter = 0;
 136
 137                 int wait_ret;
 138                 while( client->m_list->next == NULL && remaining >= 0 ) {
 139
 140                         if( (wait_ret= session_wait( client->session, remaining)) ) {
 141                                 client->error = 1;
 142                                 osrfLogDebug(OSRF_LOG_MARK, "session_wait returned failure code %d: setting error=1\n", wait_ret);
 143                                 return NULL;
 144                         }
 145
 146                         ++counter;
 147
 148 #ifdef _ROUTER
 149                         // session_wait returns -1 if there is no more data and we're a router
 150                         if( remaining == 0 ) { // && wait_ret == -1 ) {
 151                                 break;
 152                         }
 153 #else
 154                         if( remaining == 0 ) // or infinite loop
 155                                 break;
 156 #endif
 157
 158                         remaining -= (int) (time(NULL) - start);
 159                 }
 160
 161         }
 162
 163         /* again, see if there are any messages in the message queue */
 164         if( client->m_list->next != NULL ) {
 165                 /* pop off the first one... */
 166                 node = client->m_list->next;
 167                 client->m_list->next = node->next;
 168                 msg = node->message;
 169                 free( node );
 170                 return msg;
 171
 172         } else {
 173                 return NULL;
 174         }
 175 }
 176
 177 /* throw the message into the message queue */
 178 void client_message_handler( void* client, transport_message* msg ){
 179
 180         if(client == NULL) return;
 181         if(msg == NULL) return;
 182
 183         transport_client* cli = (transport_client*) client;
 184
 185         transport_message_node* node = safe_malloc( sizeof( transport_message_node) );
 186         node->next = NULL;
 187         node->type = MESSAGE_LIST_ITEM;
 188         node->message = msg;
 189
 190
 191         /* find the last node and put this onto the end */
 192         transport_message_node* tail = cli->m_list;
 193         transport_message_node* current = tail->next;
 194
 195         while( current != NULL ) {
 196                 tail = current;
 197                 current = current->next;
 198         }
 199         tail->next = node;
 200 }
 201
 202
 203 int client_free( transport_client* client ){
 204         if(client == NULL) return 0;
 205
 206         session_free( client->session );
 207         transport_message_node* current = client->m_list->next;
 208         transport_message_node* next;
 209
 210         /* deallocate the list of messages */
 211         while( current != NULL ) {
 212                 next = current->next;
 213                 message_free( current->message );
 214                 free(current);
 215                 current = next;
 216         }
 217
 218         free( client->m_list );
 219         free( client );
 220         return 1;
 221 }
 222