src/libopensrf/osrf_app_session.c

   1 /**
   2         @file osrf_app_session.c
   3         @brief Implementation of osrfAppSession.
   4 */
   5
   6 #include <time.h>
   7 #include "opensrf/osrf_app_session.h"
   8 #include "opensrf/osrf_stack.h"
   9
  10 struct osrf_app_request_struct {
  11         /** The controlling session. */
  12         struct osrf_app_session_struct* session;
  13
  14         /** Request id.  It is the same as the thread_trace of the REQUEST message
  15                 for which it was created.
  16         */
  17         int request_id;
  18         /** True if we have received a 'request complete' message from our request. */
  19         int complete;
  20         /** The original REQUEST message payload. */
  21         osrfMessage* payload;
  22         /** Linked list of responses to the request. */
  23         osrfMessage* result;
  24
  25         /** Boolean; if true, then a call that is waiting on a response will reset the
  26         timeout and set this variable back to false. */
  27         int reset_timeout;
  28 };
  29 typedef struct osrf_app_request_struct osrfAppRequest;
  30
  31 /* Send the given message */
  32 static int _osrf_app_session_send( osrfAppSession*, osrfMessage* msg );
  33
  34 static int osrfAppSessionMakeLocaleRequest(
  35                 osrfAppSession* session, const jsonObject* params, const char* method_name,
  36                 int protocol, osrfStringArray* param_strings, char* locale );
  37
  38 /** @brief The global session cache.
  39
  40         Key: session_id.  Data: osrfAppSession.
  41 */
  42 static osrfHash* osrfAppSessionCache = NULL;
  43
  44 // --------------------------------------------------------------------------
  45 // Request API
  46 // --------------------------------------------------------------------------
  47
  48 /**
  49         @brief Create a new osrfAppRequest.
  50         @param session Pointer to the osrfAppSession that will own the new osrfAppRequest.
  51         @param msg Pointer to the osrfMessage representing the request.
  52         @return Pointer to the new osrfAppRequest.
  53
  54         The calling code is responsible for freeing the osrfAppRequest by calling
  55         _osrf_app_request_free().  In practice this normally happens via a callback installed
  56         in an osrfList.
  57 */
  58 static osrfAppRequest* _osrf_app_request_init(
  59                 osrfAppSession* session, osrfMessage* msg ) {
  60
  61         osrfAppRequest* req =
  62                 (osrfAppRequest*) safe_malloc(sizeof(osrfAppRequest));
  63
  64         req->session        = session;
  65         req->request_id     = msg->thread_trace;
  66         req->complete       = 0;
  67         req->payload        = msg;
  68         req->result         = NULL;
  69         req->reset_timeout  = 0;
  70
  71         return req;
  72 }
  73
  74
  75 /**
  76         @brief Free an osrfAppRequest and everything it owns.
  77         @param req Pointer to an osrfAppRequest, cast to a void pointer.
  78
  79         This function is installed as a callback in the osrfList request_queue, a member
  80         of osrfAppSession.
  81 */
  82 static void _osrf_app_request_free( void * req ){
  83         if( req ) {
  84                 osrfAppRequest* r = (osrfAppRequest*) req;
  85                 if( r->payload )
  86                         osrfMessageFree( r->payload );
  87
  88                 /* Free the messages in the result queue */
  89                 osrfMessage* next_msg;
  90                 while( r->result ) {
  91                         next_msg = r->result->next;
  92                         osrfMessageFree( r->result );
  93                         r->result = next_msg;
  94                 }
  95
  96                 free( r );
  97         }
  98 }
  99
 100 /**
 101         @brief Append a new message to the list of responses to a request.
 102         @param req Pointer to the osrfAppRequest for the original REQUEST message.
 103         @param result Pointer to an osrfMessage received in response to the request.
 104
 105         We maintain a linked list of response messages, and traverse it to find the end.
 106 */
 107 static void _osrf_app_request_push_queue( osrfAppRequest* req, osrfMessage* result ){
 108         if(req == NULL || result == NULL)
 109                 return;
 110
 111         osrfLogDebug( OSRF_LOG_MARK, "App Session pushing request [%d] onto request queue",
 112                         result->thread_trace );
 113         if(req->result == NULL) {
 114                 req->result = result;   // Add the first node
 115
 116         } else {
 117
 118                 // Find the last node in the list, and append the new node to it
 119                 osrfMessage* ptr = req->result;
 120                 osrfMessage* ptr2 = req->result->next;
 121                 while( ptr2 ) {
 122                         ptr = ptr2;
 123                         ptr2 = ptr2->next;
 124                 }
 125                 ptr->next = result;
 126         }
 127 }
 128
 129 /**
 130         @brief Remove an osrfAppRequest (identified by request_id) from an osrfAppSession.
 131         @param session Pointer to the osrfAppSession that owns the osrfAppRequest.
 132         @param req_id request_id of the osrfAppRequest to be removed.
 133
 134         The osrfAppRequest itself is freed by a callback installed in the osrfList where it resides.
 135 */
 136 void osrf_app_session_request_finish(
 137                 osrfAppSession* session, int req_id ){
 138
 139         if( session ) {
 140                 osrfAppRequest* req = OSRF_LIST_GET_INDEX( session->request_queue, req_id );
 141                 if(req)
 142                         osrfListRemove( req->session->request_queue, req->request_id );
 143         }
 144 }
 145
 146
 147 /**
 148         @brief Set the timeout for a request to one second.
 149         @param session Pointer to the relevant osrfAppSession.
 150         @param req_id Request ID of the request whose timeout is to be reset.
 151
 152         The request to be reset is identified by the combination of session and request id.
 153 */
 154 void osrf_app_session_request_reset_timeout( osrfAppSession* session, int req_id ) {
 155         if(session == NULL)
 156                 return;
 157         osrfLogDebug( OSRF_LOG_MARK, "Resetting request timeout %d", req_id );
 158         osrfAppRequest* req = OSRF_LIST_GET_INDEX( session->request_queue, req_id );
 159         if( req )
 160                 req->reset_timeout = 1;
 161 }
 162
 163 /**
 164         Checks the receive queue for messages.  If any are found, the first
 165         is popped off and returned.  Otherwise, this method will wait at most timeout
 166         seconds for a message to appear in the receive queue.  Once it arrives it is returned.
 167         If no messages arrive in the timeout provided, null is returned.
 168 */
 169 /**
 170         @brief Return the next response message for a given request, subject to a timeout.
 171         @param req Pointer to the osrfAppRequest representing the request.
 172         @param timeout Maxmimum time to wait, in seconds.
 173
 174         @return
 175
 176         If the input queue for this request is not empty, dequeue the next message and return it.
 177 */
 178 static osrfMessage* _osrf_app_request_recv( osrfAppRequest* req, int timeout ) {
 179
 180         if(req == NULL) return NULL;
 181
 182         if( req->result != NULL ) {
 183                 /* Dequeue the next message in the list */
 184                 osrfMessage* tmp_msg = req->result;
 185                 req->result = req->result->next;
 186                 return tmp_msg;
 187         }
 188
 189         time_t start = time(NULL);
 190         time_t remaining = (time_t) timeout;
 191
 192         while( remaining >= 0 ) {
 193                 /* tell the session to wait for stuff */
 194                 osrfLogDebug( OSRF_LOG_MARK,  "In app_request receive with remaining time [%d]",
 195                                 (int) remaining );
 196
 197                 osrf_app_session_queue_wait( req->session, 0, NULL );
 198                 if(req->session->transport_error) {
 199                         osrfLogError(OSRF_LOG_MARK, "Transport error in recv()");
 200                         return NULL;
 201                 }
 202
 203                 if( req->result != NULL ) { /* if we received anything */
 204                         /* pop off the first message in the list */
 205                         osrfLogDebug( OSRF_LOG_MARK, "app_request_recv received a message, returning it" );
 206                         osrfMessage* ret_msg = req->result;
 207                         req->result = ret_msg->next;
 208                         if (ret_msg->sender_locale)
 209                                 osrf_app_session_set_locale(req->session, ret_msg->sender_locale);
 210
 211                         return ret_msg;
 212                 }
 213
 214                 if( req->complete )
 215                         return NULL;
 216
 217                 osrf_app_session_queue_wait( req->session, (int) remaining, NULL );
 218
 219                 if(req->session->transport_error) {
 220                         osrfLogError(OSRF_LOG_MARK, "Transport error in recv()");
 221                         return NULL;
 222                 }
 223
 224                 if( req->result != NULL ) { /* if we received anything */
 225                         /* pop off the first message in the list */
 226                         osrfLogDebug( OSRF_LOG_MARK,  "app_request_recv received a message, returning it");
 227                         osrfMessage* ret_msg = req->result;
 228                         req->result = ret_msg->next;
 229                         if (ret_msg->sender_locale)
 230                                 osrf_app_session_set_locale(req->session, ret_msg->sender_locale);
 231
 232                         return ret_msg;
 233                 }
 234                 if( req->complete )
 235                         return NULL;
 236
 237                 if(req->reset_timeout) {
 238                         remaining = (time_t) timeout;
 239                         req->reset_timeout = 0;
 240                         osrfLogDebug( OSRF_LOG_MARK, "Received a timeout reset");
 241                 } else {
 242                         remaining -= (int) (time(NULL) - start);
 243                 }
 244         }
 245
 246         char* paramString = jsonObjectToJSON(req->payload->_params);
 247         osrfLogInfo( OSRF_LOG_MARK, "Returning NULL from app_request_recv after timeout: %s %s",
 248                 req->payload->method_name, paramString);
 249         free(paramString);
 250
 251         return NULL;
 252 }
 253
 254 /** Resend this requests original request message */
 255 static int _osrf_app_request_resend( osrfAppRequest* req ) {
 256         if(req == NULL) return 0;
 257         if(!req->complete) {
 258                 osrfLogDebug( OSRF_LOG_MARK,  "Resending request [%d]", req->request_id );
 259                 return _osrf_app_session_send( req->session, req->payload );
 260         }
 261         return 1;
 262 }
 263
 264
 265 // --------------------------------------------------------------------------
 266 // Session API
 267 // --------------------------------------------------------------------------
 268
 269 /** Install a locale for the session */
 270 char* osrf_app_session_set_locale( osrfAppSession* session, const char* locale ) {
 271         if (!session || !locale)
 272                 return NULL;
 273
 274         if(session->session_locale) {
 275                 if( strlen(session->session_locale) >= strlen(locale) ) {
 276                         /* There's room available; just copy */
 277                         strcpy(session->session_locale, locale);
 278                 } else {
 279                         free(session->session_locale);
 280                         session->session_locale = strdup( locale );
 281                 }
 282         } else {
 283                 session->session_locale = strdup( locale );
 284         }
 285
 286         return session->session_locale;
 287 }
 288
 289 /**
 290         @brief Find the osrfAppSession for a given session id.
 291         @param session_id The session id to look for.
 292         @return Pointer to the corresponding osrfAppSession if found, or NULL if not.
 293
 294         Search the global session cache for the specified session id.
 295 */
 296 osrfAppSession* osrf_app_session_find_session( const char* session_id ) {
 297         if(session_id)
 298                 return osrfHashGet( osrfAppSessionCache, session_id );
 299         return NULL;
 300 }
 301
 302
 303 /**
 304         @brief Add a session to the global session cache, keyed by session id.
 305         @param session Pointer to the osrfAppSession to be added.
 306
 307         If a cache doesn't exist yet, create one.  It's an osrfHash using session ids for the
 308         key and osrfAppSessions for the data.
 309 */
 310 static void _osrf_app_session_push_session( osrfAppSession* session ) {
 311         if( session ) {
 312                 if( osrfAppSessionCache == NULL )
 313                         osrfAppSessionCache = osrfNewHash();
 314                 if( osrfHashGet( osrfAppSessionCache, session->session_id ) )
 315                         return;   // A session with this id is already in the cache.  Shouldn't happen.
 316                 osrfHashSet( osrfAppSessionCache, session, session->session_id );
 317         }
 318 }
 319
 320 /** Allocates and initializes a new app_session */
 321
 322 osrfAppSession* osrfAppSessionClientInit( const char* remote_service ) {
 323
 324         if (!remote_service) {
 325                 osrfLogWarning( OSRF_LOG_MARK, "No remote service specified in osrfAppSessionClientInit");
 326                 return NULL;
 327         }
 328
 329         osrfAppSession* session = safe_malloc(sizeof(osrfAppSession));
 330
 331         session->transport_handle = osrfSystemGetTransportClient();
 332         if( session->transport_handle == NULL ) {
 333                 osrfLogWarning( OSRF_LOG_MARK, "No transport client for service 'client'");
 334                 free( session );
 335                 return NULL;
 336         }
 337
 338         osrfStringArray* arr = osrfNewStringArray(8);
 339         osrfConfigGetValueList(NULL, arr, "/domain");
 340         const char* domain = osrfStringArrayGetString(arr, 0);
 341
 342         if (!domain) {
 343                 osrfLogWarning( OSRF_LOG_MARK, "No domains specified in the OpenSRF config file");
 344                 free( session );
 345                 osrfStringArrayFree(arr);
 346                 return NULL;
 347         }
 348
 349         char* router_name = osrfConfigGetValue(NULL, "/router_name");
 350         if (!router_name) {
 351                 osrfLogWarning( OSRF_LOG_MARK, "No router name specified in the OpenSRF config file");
 352                 free( session );
 353                 osrfStringArrayFree(arr);
 354                 return NULL;
 355         }
 356
 357         char target_buf[512];
 358         target_buf[ 0 ] = '\0';
 359
 360         int len = snprintf( target_buf, sizeof(target_buf), "%s@%s/%s",
 361                         router_name ? router_name : "(null)",
 362                         domain ? domain : "(null)",
 363                         remote_service ? remote_service : "(null)" );
 364         osrfStringArrayFree(arr);
 365         //free(domain);
 366         free(router_name);
 367
 368         if( len >= sizeof( target_buf ) ) {
 369                 osrfLogWarning( OSRF_LOG_MARK, "Buffer overflow for remote_id");
 370                 free( session );
 371                 return NULL;
 372         }
 373
 374         session->request_queue = osrfNewList();
 375         session->request_queue->freeItem = &_osrf_app_request_free;
 376         session->remote_id = strdup(target_buf);
 377         session->orig_remote_id = strdup(session->remote_id);
 378         session->remote_service = strdup(remote_service);
 379         session->session_locale = NULL;
 380         session->transport_error = 0;
 381
 382         #ifdef ASSUME_STATELESS
 383         session->stateless = 1;
 384         osrfLogDebug( OSRF_LOG_MARK, "%s session is stateless", remote_service );
 385         #else
 386         session->stateless = 0;
 387         osrfLogDebug( OSRF_LOG_MARK, "%s session is NOT stateless", remote_service );
 388         #endif
 389
 390         /* build a chunky, random session id */
 391         char id[256];
 392
 393         snprintf(id, sizeof(id), "%f.%d%ld", get_timestamp_millis(), (int)time(NULL), (long) getpid());
 394         session->session_id = strdup(id);
 395         osrfLogDebug( OSRF_LOG_MARK,  "Building a new client session with id [%s] [%s]",
 396                         session->remote_service, session->session_id );
 397
 398         session->thread_trace = 0;
 399         session->state = OSRF_SESSION_DISCONNECTED;
 400         session->type = OSRF_SESSION_CLIENT;
 401
 402         session->userData = NULL;
 403         session->userDataFree = NULL;
 404
 405         _osrf_app_session_push_session( session );
 406         return session;
 407 }
 408
 409 osrfAppSession* osrf_app_server_session_init(
 410                 const char* session_id, const char* our_app, const char* remote_id ) {
 411
 412         osrfLogDebug( OSRF_LOG_MARK, "Initing server session with session id %s, service %s,"
 413                         " and remote_id %s", session_id, our_app, remote_id );
 414
 415         osrfAppSession* session = osrf_app_session_find_session( session_id );
 416         if(session) return session;
 417
 418         session = safe_malloc(sizeof(osrfAppSession));
 419
 420         session->transport_handle = osrfSystemGetTransportClient();
 421         if( session->transport_handle == NULL ) {
 422                 osrfLogWarning( OSRF_LOG_MARK, "No transport client for service '%s'", our_app );
 423                 free(session);
 424                 return NULL;
 425         }
 426
 427         int stateless = 0;
 428         char* statel = osrf_settings_host_value("/apps/%s/stateless", our_app );
 429         if(statel) stateless = atoi(statel);
 430         free(statel);
 431
 432
 433         session->request_queue = osrfNewList();
 434         session->request_queue->freeItem = &_osrf_app_request_free;
 435         session->remote_id = strdup(remote_id);
 436         session->orig_remote_id = strdup(remote_id);
 437         session->session_id = strdup(session_id);
 438         session->remote_service = strdup(our_app);
 439         session->stateless = stateless;
 440
 441         #ifdef ASSUME_STATELESS
 442         session->stateless = 1;
 443         #else
 444         session->stateless = 0;
 445         #endif
 446
 447         session->thread_trace = 0;
 448         session->state = OSRF_SESSION_DISCONNECTED;
 449         session->type = OSRF_SESSION_SERVER;
 450         session->session_locale = NULL;
 451
 452         session->userData = NULL;
 453         session->userDataFree = NULL;
 454
 455         _osrf_app_session_push_session( session );
 456         return session;
 457
 458 }
 459
 460 /**
 461         @brief Create a REQUEST message, send it, and save it for future reference.
 462         @param session Pointer to the current session, which has the addressing information.
 463         @param params One way of specifying the parameters for the method.
 464         @param method_name The name of the method to be called.
 465         @param protocol Protocol.
 466         @param param_strings Another way of specifying the parameters for the method.
 467         @return The request ID of the resulting REQUEST message, or -1 upon error.
 468
 469         DEPRECATED.  Use osrfAppSessionSendRequest() instead.  It is identical except that it
 470         doesn't use the param_strings argument, which is redundant, confusing, and unused.
 471
 472         If @a params is non-NULL, use it to specify the parameters to the method.  Otherwise
 473         use @a param_strings.
 474
 475         If @a params points to a JSON_ARRAY, then pass each element of the array as a separate
 476         parameter.  If @a params points to any other kind of jsonObject, pass it as a single
 477         parameter.
 478
 479         If @a params is NULL, and @a param_strings is not NULL, then each pointer in the
 480         osrfStringArray must point to a JSON string encoding a parameter.  Pass them.
 481
 482         At this writing, all calls to this function use @a params to pass parameters, rather than
 483         @a param_strings.
 484
 485         This function is a thin wrapper for osrfAppSessionMakeLocaleRequest().
 486 */
 487 int osrfAppSessionMakeRequest(
 488                 osrfAppSession* session, const jsonObject* params,
 489                 const char* method_name, int protocol, osrfStringArray* param_strings ) {
 490
 491         osrfLogWarning( OSRF_LOG_MARK, "Function osrfAppSessionMakeRequest() is deprecasted; "
 492                         "call osrfAppSessionSendRequest() instead" );
 493         return osrfAppSessionMakeLocaleRequest( session, params,
 494                         method_name, protocol, param_strings, NULL );
 495 }
 496
 497 /**
 498         @brief Create a REQUEST message, send it, and save it for future reference.
 499         @param session Pointer to the current session, which has the addressing information.
 500         @param params One way of specifying the parameters for the method.
 501         @param method_name The name of the method to be called.
 502         @param protocol Protocol.
 503         @return The request ID of the resulting REQUEST message, or -1 upon error.
 504
 505         If @a params points to a JSON_ARRAY, then pass each element of the array as a separate
 506         parameter.  If @a params points to any other kind of jsonObject, pass it as a single
 507         parameter.
 508
 509         This function is a thin wrapper for osrfAppSessionMakeLocaleRequest().
 510 */
 511 int osrfAppSessionSendRequest( osrfAppSession* session, const jsonObject* params,
 512                 const char* method_name, int protocol ) {
 513
 514         return osrfAppSessionMakeLocaleRequest( session, params,
 515                  method_name, protocol, NULL, NULL );
 516 }
 517
 518 /**
 519         @brief Create a REQUEST message, send it, and save it for future reference.
 520         @param session Pointer to the current session, which has the addressing information.
 521         @param params One way of specifying the parameters for the method.
 522         @param method_name The name of the method to be called.
 523         @param protocol Protocol.
 524         @param param_strings Another way of specifying the parameters for the method.
 525         @param locale Pointer to a locale string.
 526         @return The request ID of the resulting REQUEST message, or -1 upon error.
 527
 528         See the discussion of osrfAppSessionSendRequest(), which at this writing is the only
 529         place that calls this function, except for the similar but deprecated function
 530         osrfAppSessionMakeRequest().
 531
 532         At this writing, the @a param_strings and @a locale parameters are always NULL.
 533 */
 534 static int osrfAppSessionMakeLocaleRequest(
 535                 osrfAppSession* session, const jsonObject* params, const char* method_name,
 536                 int protocol, osrfStringArray* param_strings, char* locale ) {
 537
 538         if(session == NULL) return -1;
 539
 540         osrfLogMkXid();
 541
 542         osrfMessage* req_msg = osrf_message_init( REQUEST, ++(session->thread_trace), protocol );
 543         osrf_message_set_method(req_msg, method_name);
 544
 545         if (locale) {
 546                 osrf_message_set_locale(req_msg, locale);
 547         } else if (session->session_locale) {
 548                 osrf_message_set_locale(req_msg, session->session_locale);
 549         }
 550
 551         if(params) {
 552                 osrf_message_set_params(req_msg, params);
 553
 554         } else {
 555
 556                 if(param_strings) {
 557                         int i;
 558                         for(i = 0; i!= param_strings->size ; i++ ) {
 559                                 osrf_message_add_param(req_msg,
 560                                         osrfStringArrayGetString(param_strings,i));
 561                         }
 562                 }
 563         }
 564
 565         osrfAppRequest* req = _osrf_app_request_init( session, req_msg );
 566         if(_osrf_app_session_send( session, req_msg ) ) {
 567                 osrfLogWarning( OSRF_LOG_MARK,  "Error sending request message [%d]",
 568                                 session->thread_trace );
 569                 _osrf_app_request_free(req);
 570                 return -1;
 571         }
 572
 573         osrfLogDebug( OSRF_LOG_MARK,  "Pushing [%d] onto request queue for session [%s] [%s]",
 574                         req->request_id, session->remote_service, session->session_id );
 575         osrfListSet( session->request_queue, req, req->request_id );
 576         return req->request_id;
 577 }
 578
 579 void osrf_app_session_set_complete( osrfAppSession* session, int request_id ) {
 580         if(session == NULL)
 581                 return;
 582
 583         osrfAppRequest* req = OSRF_LIST_GET_INDEX( session->request_queue, request_id );
 584         if(req) req->complete = 1;
 585 }
 586
 587 int osrf_app_session_request_complete( const osrfAppSession* session, int request_id ) {
 588         if(session == NULL)
 589                 return 0;
 590         osrfAppRequest* req = OSRF_LIST_GET_INDEX( session->request_queue, request_id );
 591         if(req)
 592                 return req->complete;
 593         return 0;
 594 }
 595
 596
 597 /** Resets the remote connection id to that of the original*/
 598 void osrf_app_session_reset_remote( osrfAppSession* session ){
 599         if( session==NULL )
 600                 return;
 601
 602         osrfLogDebug( OSRF_LOG_MARK,  "App Session [%s] [%s] resetting remote id to %s",
 603                         session->remote_service, session->session_id, session->orig_remote_id );
 604
 605         osrf_app_session_set_remote( session, session->orig_remote_id );
 606 }
 607
 608 void osrf_app_session_set_remote( osrfAppSession* session, const char* remote_id ) {
 609         if(session == NULL)
 610                 return;
 611
 612         if( session->remote_id ) {
 613                 if( strlen(session->remote_id) >= strlen(remote_id) ) {
 614                         // There's enough room; just copy it
 615                         strcpy(session->remote_id, remote_id);
 616                 } else {
 617                         free(session->remote_id );
 618                         session->remote_id = strdup( remote_id );
 619                 }
 620         } else
 621                 session->remote_id = strdup( remote_id );
 622 }
 623
 624 /**
 625         pushes the given message into the result list of the app_request
 626         with the given request_id
 627 */
 628 int osrf_app_session_push_queue(
 629                 osrfAppSession* session, osrfMessage* msg ){
 630         if(session == NULL || msg == NULL) return 0;
 631
 632         osrfAppRequest* req = OSRF_LIST_GET_INDEX( session->request_queue, msg->thread_trace );
 633         if(req == NULL) return 0;
 634         _osrf_app_request_push_queue( req, msg );
 635
 636         return 0;
 637 }
 638
 639 /** Attempts to connect to the remote service */
 640 int osrfAppSessionConnect( osrfAppSession* session ) {
 641
 642         if(session == NULL)
 643                 return 0;
 644
 645         if(session->state == OSRF_SESSION_CONNECTED) {
 646                 return 1;
 647         }
 648
 649         int timeout = 5; /* XXX CONFIG VALUE */
 650
 651         osrfLogDebug( OSRF_LOG_MARK,  "AppSession connecting to %s", session->remote_id );
 652
 653         /* defaulting to protocol 1 for now */
 654         osrfMessage* con_msg = osrf_message_init( CONNECT, session->thread_trace, 1 );
 655         osrf_app_session_reset_remote( session );
 656         session->state = OSRF_SESSION_CONNECTING;
 657         int ret = _osrf_app_session_send( session, con_msg );
 658         osrfMessageFree(con_msg);
 659         if(ret)
 660                 return 0;
 661
 662         time_t start = time(NULL);
 663         time_t remaining = (time_t) timeout;
 664
 665         while( session->state != OSRF_SESSION_CONNECTED && remaining >= 0 ) {
 666                 osrf_app_session_queue_wait( session, remaining, NULL );
 667                 if(session->transport_error) {
 668                         osrfLogError(OSRF_LOG_MARK, "cannot communicate with %s", session->remote_service);
 669                         return 0;
 670                 }
 671                 remaining -= (int) (time(NULL) - start);
 672         }
 673
 674         if(session->state == OSRF_SESSION_CONNECTED)
 675                 osrfLogDebug( OSRF_LOG_MARK, " * Connected Successfully to %s", session->remote_service );
 676
 677         if(session->state != OSRF_SESSION_CONNECTED)
 678                 return 0;
 679
 680         return 1;
 681 }
 682
 683
 684
 685 /** Disconnects from the remote service */
 686 int osrf_app_session_disconnect( osrfAppSession* session){
 687         if(session == NULL)
 688                 return 1;
 689
 690         if(session->state == OSRF_SESSION_DISCONNECTED)
 691                 return 1;
 692
 693         if(session->stateless && session->state != OSRF_SESSION_CONNECTED) {
 694                 osrfLogDebug( OSRF_LOG_MARK,
 695                                 "Exiting disconnect on stateless session %s",
 696                                 session->session_id);
 697                 return 1;
 698         }
 699
 700         osrfLogDebug(OSRF_LOG_MARK,  "AppSession disconnecting from %s", session->remote_id );
 701
 702         osrfMessage* dis_msg = osrf_message_init( DISCONNECT, session->thread_trace, 1 );
 703         _osrf_app_session_send( session, dis_msg );
 704         session->state = OSRF_SESSION_DISCONNECTED;
 705
 706         osrfMessageFree( dis_msg );
 707         osrf_app_session_reset_remote( session );
 708         return 1;
 709 }
 710
 711 int osrf_app_session_request_resend( osrfAppSession* session, int req_id ) {
 712         osrfAppRequest* req = OSRF_LIST_GET_INDEX( session->request_queue, req_id );
 713         return _osrf_app_request_resend( req );
 714 }
 715
 716
 717 static int osrfAppSessionSendBatch( osrfAppSession* session, osrfMessage* msgs[], int size ) {
 718
 719         if( !(session && msgs && size > 0) ) return 0;
 720         int retval = 0;
 721
 722         osrfMessage* msg = msgs[0];
 723
 724         if(msg) {
 725
 726                 osrf_app_session_queue_wait( session, 0, NULL );
 727
 728                 if(session->state != OSRF_SESSION_CONNECTED)  {
 729
 730                         if(session->stateless) { /* stateless session always send to the root listener */
 731                                 osrf_app_session_reset_remote(session);
 732
 733                         } else {
 734
 735                                 /* do an auto-connect if necessary */
 736                                 if( ! session->stateless &&
 737                                         (msg->m_type != CONNECT) &&
 738                                         (msg->m_type != DISCONNECT) &&
 739                                         (session->state != OSRF_SESSION_CONNECTED) ) {
 740
 741                                         if(!osrfAppSessionConnect( session ))
 742                                                 return 0;
 743                                 }
 744                         }
 745                 }
 746         }
 747
 748         char* string = osrfMessageSerializeBatch(msgs, size);
 749
 750         if( string ) {
 751
 752                 transport_message* t_msg = message_init(
 753                                 string, "", session->session_id, session->remote_id, NULL );
 754                 message_set_osrf_xid( t_msg, osrfLogGetXid() );
 755
 756                 retval = client_send_message( session->transport_handle, t_msg );
 757
 758                 if( retval ) osrfLogError(OSRF_LOG_MARK, "client_send_message failed");
 759
 760                 osrfLogInfo(OSRF_LOG_MARK, "[%s] sent %d bytes of data to %s",
 761                         session->remote_service, strlen(string), t_msg->recipient );
 762
 763                 osrfLogDebug(OSRF_LOG_MARK, "Sent: %s", string );
 764
 765                 free(string);
 766                 message_free( t_msg );
 767         }
 768
 769         return retval;
 770 }
 771
 772
 773
 774 static int _osrf_app_session_send( osrfAppSession* session, osrfMessage* msg ){
 775         if( !(session && msg) ) return 0;
 776         osrfMessage* a[1];
 777         a[0] = msg;
 778         return osrfAppSessionSendBatch( session, a, 1 );
 779 }
 780
 781
 782 /**
 783         Waits up to 'timeout' seconds for some data to arrive.
 784         Any data that arrives will be processed according to its
 785         payload and message type.  This method will return after
 786         any data has arrived.
 787 */
 788 int osrf_app_session_queue_wait( osrfAppSession* session, int timeout, int* recvd ){
 789         if(session == NULL) return 0;
 790         osrfLogDebug(OSRF_LOG_MARK, "AppSession in queue_wait with timeout %d", timeout );
 791         return osrf_stack_process(session->transport_handle, timeout, recvd);
 792 }
 793
 794 /** Disconnects (if client) and removes the given session from the global session cache
 795         ! This frees all attached app_requests !
 796 */
 797 void osrfAppSessionFree( osrfAppSession* session ){
 798         if(session == NULL) return;
 799
 800         /* Disconnect */
 801
 802         osrfLogDebug(OSRF_LOG_MARK,  "AppSession [%s] [%s] destroying self and deleting requests",
 803                         session->remote_service, session->session_id );
 804         if(session->type == OSRF_SESSION_CLIENT
 805                         && session->state != OSRF_SESSION_DISCONNECTED ) { /* disconnect if we're a client */
 806                 osrfMessage* dis_msg = osrf_message_init( DISCONNECT, session->thread_trace, 1 );
 807                 _osrf_app_session_send( session, dis_msg );
 808                 osrfMessageFree(dis_msg);
 809         }
 810
 811         /* Remove self from the global session cache */
 812
 813         osrfHashRemove( osrfAppSessionCache, session->session_id );
 814
 815         /* Free the memory */
 816
 817         if( session->userDataFree && session->userData )
 818                 session->userDataFree(session->userData);
 819
 820         if(session->session_locale)
 821                 free(session->session_locale);
 822
 823         free(session->remote_id);
 824         free(session->orig_remote_id);
 825         free(session->session_id);
 826         free(session->remote_service);
 827         osrfListFree(session->request_queue);
 828         free(session);
 829 }
 830
 831 osrfMessage* osrfAppSessionRequestRecv(
 832                 osrfAppSession* session, int req_id, int timeout ) {
 833         if(req_id < 0 || session == NULL)
 834                 return NULL;
 835         osrfAppRequest* req = OSRF_LIST_GET_INDEX( session->request_queue, req_id );
 836         return _osrf_app_request_recv( req, timeout );
 837 }
 838
 839
 840
 841 int osrfAppRequestRespond( osrfAppSession* ses, int requestId, const jsonObject* data ) {
 842         if(!ses || ! data ) return -1;
 843
 844         osrfMessage* msg = osrf_message_init( RESULT, requestId, 1 );
 845         osrf_message_set_status_info( msg, NULL, "OK", OSRF_STATUS_OK );
 846         char* json = jsonObjectToJSON( data );
 847
 848         osrf_message_set_result_content( msg, json );
 849         _osrf_app_session_send( ses, msg );
 850
 851         free(json);
 852         osrfMessageFree( msg );
 853
 854         return 0;
 855 }
 856
 857
 858 int osrfAppRequestRespondComplete(
 859                 osrfAppSession* ses, int requestId, const jsonObject* data ) {
 860
 861         osrfMessage* status = osrf_message_init( STATUS, requestId, 1);
 862         osrf_message_set_status_info( status, "osrfConnectStatus", "Request Complete",
 863                         OSRF_STATUS_COMPLETE );
 864
 865         if (data) {
 866                 osrfMessage* payload = osrf_message_init( RESULT, requestId, 1 );
 867                 osrf_message_set_status_info( payload, NULL, "OK", OSRF_STATUS_OK );
 868
 869                 char* json = jsonObjectToJSON( data );
 870                 osrf_message_set_result_content( payload, json );
 871                 free(json);
 872
 873                 osrfMessage* ms[2];
 874                 ms[0] = payload;
 875                 ms[1] = status;
 876
 877                 osrfAppSessionSendBatch( ses, ms, 2 );
 878
 879                 osrfMessageFree( payload );
 880         } else {
 881                 osrfAppSessionSendBatch( ses, &status, 1 );
 882         }
 883
 884         osrfMessageFree( status );
 885
 886         return 0;
 887 }
 888
 889 int osrfAppSessionStatus( osrfAppSession* ses, int type,
 890                 const char* name, int reqId, const char* message ) {
 891
 892         if(ses) {
 893                 osrfMessage* msg = osrf_message_init( STATUS, reqId, 1);
 894                 osrf_message_set_status_info( msg, name, message, type );
 895                 _osrf_app_session_send( ses, msg );
 896                 osrfMessageFree( msg );
 897                 return 0;
 898         }
 899         return -1;
 900 }
 901
 902 /**
 903         @brief Free the global session cache.
 904
 905         Note that the osrfHash that implements the global session cache does @em not have a
 906         callback function installed for freeing its cargo.  As a result, any outstanding
 907         osrfAppSessions are leaked, along with all the osrfAppRequests and osrfMessages they
 908         own.
 909 */
 910 void osrfAppSessionCleanup( void ) {
 911         osrfHashFree(osrfAppSessionCache);
 912         osrfAppSessionCache = NULL;
 913 }
 914
 915
 916
 917
 918