src/libopensrf/transport_client.c

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