7681a8702b4e73068dd6fbf1a10b9fbac596d565
[OpenSRF.git] / src / libopensrf / osrf_app_session.c
1 #include <opensrf/osrf_app_session.h>
2 #include <time.h>
3
4 struct osrf_app_request_struct {
5         /** Our controlling session */
6         struct osrf_app_session_struct* session;
7
8         /** our "id" */
9         int request_id;
10         /** True if we have received a 'request complete' message from our request */
11         int complete;
12         /** Our original request payload */
13         osrfMessage* payload;
14         /** List of responses to our request */
15         osrfMessage* result;
16
17         /* if set to true, then a call that is waiting on a response, will reset the 
18         timeout and set this variable back to false */
19         int reset_timeout;
20 };
21 typedef struct osrf_app_request_struct osrfAppRequest;
22
23 /** Send the given message */
24 static int _osrf_app_session_send( osrfAppSession*, osrfMessage* msg );
25
26 static int osrfAppSessionMakeLocaleRequest(
27                 osrfAppSession* session, const jsonObject* params, const char* method_name,
28                 int protocol, osrfStringArray* param_strings, char* locale );
29
30 /* the global app_session cache */
31 static osrfHash* osrfAppSessionCache = NULL;
32
33 // --------------------------------------------------------------------------
34 // --------------------------------------------------------------------------
35 // Request API
36 // --------------------------------------------------------------------------
37
38 /** Allocates and initializes a new app_request object */
39 static osrfAppRequest* _osrf_app_request_init(
40                 osrfAppSession* session, osrfMessage* msg ) {
41
42         osrfAppRequest* req =
43                 (osrfAppRequest*) safe_malloc(sizeof(osrfAppRequest));
44
45         req->session            = session;
46         req->request_id = msg->thread_trace;
47         req->complete           = 0;
48         req->payload            = msg;
49         req->result                     = NULL;
50         req->reset_timeout  = 0;
51
52         return req;
53
54 }
55
56
57 void osrfAppSessionCleanup() {
58         osrfHashFree(osrfAppSessionCache);
59         osrfAppSessionCache = NULL;     
60 }
61
62 /** Frees memory used by an app_request object */
63 static void _osrf_app_request_free( void * req ){
64         if( req == NULL ) return;
65         osrfAppRequest* r = (osrfAppRequest*) req;
66         if( r->payload ) osrfMessageFree( r->payload );
67
68         /* Free the messages in the result queue */
69
70         osrfMessage* next_msg;
71         while( r->result ) {
72                 next_msg = r->result->next;
73                 osrfMessageFree( r->result );
74                 r->result = next_msg;
75         }
76         
77         free( r );
78 }
79
80 /** Pushes the given message onto the list of 'responses' to this request */
81 static void _osrf_app_request_push_queue( osrfAppRequest* req, osrfMessage* result ){
82         if(req == NULL || result == NULL) return;
83         osrfLogDebug( OSRF_LOG_MARK,  "App Session pushing request [%d] onto request queue", result->thread_trace );
84         if(req->result == NULL) {
85                 req->result = result;
86
87         } else {
88                 
89                 osrfMessage* ptr = req->result;
90                 osrfMessage* ptr2 = req->result->next;
91                 while( ptr2 ) {
92                         ptr = ptr2;
93                         ptr2 = ptr2->next;
94                 }
95                 ptr->next = result;
96         }
97 }
98
99 /** Removes this app_request from our session request set */
100 void osrf_app_session_request_finish( 
101                 osrfAppSession* session, int req_id ){
102
103         if(session == NULL) return;
104         osrfAppRequest* req = OSRF_LIST_GET_INDEX( session->request_queue, req_id );
105         if(req == NULL) return;
106         osrfListRemove( req->session->request_queue, req->request_id );
107 }
108
109
110 void osrf_app_session_request_reset_timeout( osrfAppSession* session, int req_id ) {
111         if(session == NULL) return;
112         osrfLogDebug( OSRF_LOG_MARK, "Resetting request timeout %d", req_id );
113         osrfAppRequest* req = OSRF_LIST_GET_INDEX( session->request_queue, req_id );
114         if(req == NULL) return;
115         req->reset_timeout = 1;
116 }
117
118 /** Checks the receive queue for messages.  If any are found, the first
119   * is popped off and returned.  Otherwise, this method will wait at most timeout 
120   * seconds for a message to appear in the receive queue.  Once it arrives it is returned.
121   * If no messages arrive in the timeout provided, null is returned.
122   */
123 static osrfMessage* _osrf_app_request_recv( osrfAppRequest* req, int timeout ) {
124
125         if(req == NULL) return NULL;
126
127         if( req->result != NULL ) {
128                 /* pop off the first message in the list */
129                 osrfMessage* tmp_msg = req->result;
130                 req->result = req->result->next;
131                 return tmp_msg;
132         }
133
134         time_t start = time(NULL);      
135         time_t remaining = (time_t) timeout;
136
137         while( remaining >= 0 ) {
138                 /* tell the session to wait for stuff */
139                 osrfLogDebug( OSRF_LOG_MARK,  "In app_request receive with remaining time [%d]", (int) remaining );
140
141                 osrf_app_session_queue_wait( req->session, 0, NULL );
142         if(req->session->transport_error) {
143             osrfLogError(OSRF_LOG_MARK, "Transport error in recv()");
144             return NULL;
145         }
146
147                 if( req->result != NULL ) { /* if we received anything */
148                         /* pop off the first message in the list */
149                         osrfLogDebug( OSRF_LOG_MARK,  "app_request_recv received a message, returning it");
150                         osrfMessage* ret_msg = req->result;
151                         req->result = ret_msg->next;
152                         if (ret_msg->sender_locale)
153                                 osrf_app_session_set_locale(req->session, ret_msg->sender_locale);
154
155                         return ret_msg;
156                 }
157
158                 if( req->complete )
159                         return NULL;
160
161                 osrf_app_session_queue_wait( req->session, (int) remaining, NULL );
162
163         if(req->session->transport_error) {
164             osrfLogError(OSRF_LOG_MARK, "Transport error in recv()");
165             return NULL;
166         }
167
168                 if( req->result != NULL ) { /* if we received anything */
169                         /* pop off the first message in the list */
170                         osrfLogDebug( OSRF_LOG_MARK,  "app_request_recv received a message, returning it");
171                         osrfMessage* ret_msg = req->result;
172                         req->result = ret_msg->next;
173                         if (ret_msg->sender_locale)
174                                 osrf_app_session_set_locale(req->session, ret_msg->sender_locale);
175
176                         return ret_msg;
177                 }
178                 if( req->complete )
179                         return NULL;
180
181                 if(req->reset_timeout) {
182                         remaining = (time_t) timeout;
183                         req->reset_timeout = 0;
184                         osrfLogDebug( OSRF_LOG_MARK, "Received a timeout reset");
185                 } else {
186                         remaining -= (int) (time(NULL) - start);
187                 }
188         }
189
190     char* paramString = jsonObjectToJSON(req->payload->_params);
191         osrfLogInfo( OSRF_LOG_MARK, "Returning NULL from app_request_recv after timeout: %s %s",
192         req->payload->method_name, paramString);
193     free(paramString);
194
195         return NULL;
196 }
197
198 /** Resend this requests original request message */
199 static int _osrf_app_request_resend( osrfAppRequest* req ) {
200         if(req == NULL) return 0;
201         if(!req->complete) {
202                 osrfLogDebug( OSRF_LOG_MARK,  "Resending request [%d]", req->request_id );
203                 return _osrf_app_session_send( req->session, req->payload );
204         }
205         return 1;
206 }
207
208
209
210 // --------------------------------------------------------------------------
211 // --------------------------------------------------------------------------
212 // Session API
213 // --------------------------------------------------------------------------
214
215 /** Install a locale for the session */
216 char* osrf_app_session_set_locale( osrfAppSession* session, const char* locale ) {
217         if (!session || !locale)
218                 return NULL;
219
220         if(session->session_locale) {
221                 if( strlen(session->session_locale) >= strlen(locale) ) {
222                         /* There's room available; just copy */
223                         strcpy(session->session_locale, locale);
224                 } else {
225                         free(session->session_locale);
226                         session->session_locale = strdup( locale );
227                 }
228         } else {
229                 session->session_locale = strdup( locale );
230         }
231
232         return session->session_locale;
233 }
234
235 /** returns a session from the global session hash */
236 osrfAppSession* osrf_app_session_find_session( const char* session_id ) {
237         if(session_id) return osrfHashGet(osrfAppSessionCache, session_id);
238         return NULL;
239 }
240
241
242 /** adds a session to the global session cache */
243 static void _osrf_app_session_push_session( osrfAppSession* session ) {
244         if(!session) return;
245         if( osrfAppSessionCache == NULL ) osrfAppSessionCache = osrfNewHash();
246         if( osrfHashGet( osrfAppSessionCache, session->session_id ) ) return;
247         osrfHashSet( osrfAppSessionCache, session, session->session_id );
248 }
249
250 /** Allocates and initializes a new app_session */
251
252 osrfAppSession* osrfAppSessionClientInit( const char* remote_service ) {
253
254         if (!remote_service) {
255                 osrfLogWarning( OSRF_LOG_MARK, "No remote service specified in osrfAppSessionClientInit");
256                 return NULL;
257         }
258
259         osrfAppSession* session = safe_malloc(sizeof(osrfAppSession));
260
261         session->transport_handle = osrfSystemGetTransportClient();
262         if( session->transport_handle == NULL ) {
263                 osrfLogWarning( OSRF_LOG_MARK, "No transport client for service 'client'");
264                 free( session );
265                 return NULL;
266         }
267
268         osrfStringArray* arr = osrfNewStringArray(8);
269         osrfConfigGetValueList(NULL, arr, "/domain");
270         char* domain = osrfStringArrayGetString(arr, 0);
271
272         if (!domain) {
273                 osrfLogWarning( OSRF_LOG_MARK, "No domains specified in the OpenSRF config file");
274                 free( session );
275                 osrfStringArrayFree(arr);
276                 return NULL;
277         }
278
279         char* router_name = osrfConfigGetValue(NULL, "/router_name");
280         if (!router_name) {
281                 osrfLogWarning( OSRF_LOG_MARK, "No router name specified in the OpenSRF config file");
282                 free( session );
283                 osrfStringArrayFree(arr);
284                 return NULL;
285         }
286
287         char target_buf[512];
288         target_buf[ 0 ] = '\0';
289
290         int len = snprintf( target_buf, sizeof(target_buf), "%s@%s/%s",
291                         router_name ? router_name : "(null)",
292                         domain ? domain : "(null)",
293                         remote_service ? remote_service : "(null)" );
294         osrfStringArrayFree(arr);
295         //free(domain);
296         free(router_name);
297
298         if( len >= sizeof( target_buf ) ) {
299                 osrfLogWarning( OSRF_LOG_MARK, "Buffer overflow for remote_id");
300                 free( session );
301                 return NULL;
302         }
303
304         session->request_queue = osrfNewList();
305         session->request_queue->freeItem = &_osrf_app_request_free;
306         session->remote_id = strdup(target_buf);
307         session->orig_remote_id = strdup(session->remote_id);
308         session->remote_service = strdup(remote_service);
309         session->session_locale = NULL;
310     session->transport_error = 0;
311
312         #ifdef ASSUME_STATELESS
313         session->stateless = 1;
314         osrfLogDebug( OSRF_LOG_MARK, "%s session is stateless", remote_service );
315         #else
316         session->stateless = 0;
317         osrfLogDebug( OSRF_LOG_MARK, "%s session is NOT stateless", remote_service );
318         #endif
319
320         /* build a chunky, random session id */
321         char id[256];
322
323         snprintf(id, sizeof(id), "%f.%d%ld", get_timestamp_millis(), (int)time(NULL), (long) getpid());
324         session->session_id = strdup(id);
325         osrfLogDebug( OSRF_LOG_MARK,  "Building a new client session with id [%s] [%s]", 
326                         session->remote_service, session->session_id );
327
328         session->thread_trace = 0;
329         session->state = OSRF_SESSION_DISCONNECTED;
330         session->type = OSRF_SESSION_CLIENT;
331         //session->next = NULL;
332
333         session->userData = NULL;
334         session->userDataFree = NULL;
335         
336         _osrf_app_session_push_session( session );
337         return session;
338 }
339
340 osrfAppSession* osrf_app_server_session_init(
341                 const char* session_id, const char* our_app, const char* remote_id ) {
342
343         osrfLogDebug( OSRF_LOG_MARK, "Initing server session with session id %s, service %s,"
344                         " and remote_id %s", session_id, our_app, remote_id );
345
346         osrfAppSession* session = osrf_app_session_find_session( session_id );
347         if(session) return session;
348
349         session = safe_malloc(sizeof(osrfAppSession));
350
351         session->transport_handle = osrfSystemGetTransportClient();
352         if( session->transport_handle == NULL ) {
353                 osrfLogWarning( OSRF_LOG_MARK, "No transport client for service '%s'", our_app );
354                 free(session);
355                 return NULL;
356         }
357
358         int stateless = 0;
359         char* statel = osrf_settings_host_value("/apps/%s/stateless", our_app );
360         if(statel) stateless = atoi(statel);
361         free(statel);
362
363
364         session->request_queue = osrfNewList();
365         session->request_queue->freeItem = &_osrf_app_request_free;
366         session->remote_id = strdup(remote_id);
367         session->orig_remote_id = strdup(remote_id);
368         session->session_id = strdup(session_id);
369         session->remote_service = strdup(our_app);
370         session->stateless = stateless;
371
372         #ifdef ASSUME_STATELESS
373         session->stateless = 1;
374         #else
375         session->stateless = 0;
376         #endif
377
378         session->thread_trace = 0;
379         session->state = OSRF_SESSION_DISCONNECTED;
380         session->type = OSRF_SESSION_SERVER;
381         session->session_locale = NULL;
382
383         session->userData = NULL;
384         session->userDataFree = NULL;
385         
386         _osrf_app_session_push_session( session );
387         return session;
388
389 }
390
391 int osrfAppSessionMakeRequest(
392                 osrfAppSession* session, const jsonObject* params,
393                 const char* method_name, int protocol, osrfStringArray* param_strings ) {
394
395         return osrfAppSessionMakeLocaleRequest( session, params,
396                         method_name, protocol, param_strings, NULL );
397 }
398
399 static int osrfAppSessionMakeLocaleRequest(
400                 osrfAppSession* session, const jsonObject* params, const char* method_name,
401                 int protocol, osrfStringArray* param_strings, char* locale ) {
402
403         if(session == NULL) return -1;
404
405         osrfLogMkXid();
406
407         osrfMessage* req_msg = osrf_message_init( REQUEST, ++(session->thread_trace), protocol );
408         osrf_message_set_method(req_msg, method_name);
409
410         if (locale) {
411                 osrf_message_set_locale(req_msg, locale);
412         } else if (session->session_locale) {
413                 osrf_message_set_locale(req_msg, session->session_locale);
414         }
415
416         if(params) {
417                 osrf_message_set_params(req_msg, params);
418
419         } else {
420
421                 if(param_strings) {
422                         int i;
423                         for(i = 0; i!= param_strings->size ; i++ ) {
424                                 osrf_message_add_param(req_msg,
425                                         osrfStringArrayGetString(param_strings,i));
426                         }
427                 }
428         }
429
430         osrfAppRequest* req = _osrf_app_request_init( session, req_msg );
431         if(_osrf_app_session_send( session, req_msg ) ) {
432                 osrfLogWarning( OSRF_LOG_MARK,  "Error sending request message [%d]", session->thread_trace );
433                 _osrf_app_request_free(req);
434                 return -1;
435         }
436
437         osrfLogDebug( OSRF_LOG_MARK,  "Pushing [%d] onto request queue for session [%s] [%s]",
438                         req->request_id, session->remote_service, session->session_id );
439         osrfListSet( session->request_queue, req, req->request_id ); 
440         return req->request_id;
441 }
442
443 void osrf_app_session_set_complete( osrfAppSession* session, int request_id ) {
444         if(session == NULL)
445                 return;
446
447         osrfAppRequest* req = OSRF_LIST_GET_INDEX( session->request_queue, request_id );
448         if(req) req->complete = 1;
449 }
450
451 int osrf_app_session_request_complete( const osrfAppSession* session, int request_id ) {
452         if(session == NULL)
453                 return 0;
454         osrfAppRequest* req = OSRF_LIST_GET_INDEX( session->request_queue, request_id );
455         if(req)
456                 return req->complete;
457         return 0;
458 }
459
460
461 /** Resets the remote connection id to that of the original*/
462 void osrf_app_session_reset_remote( osrfAppSession* session ){
463         if( session==NULL )
464                 return;
465
466         osrfLogDebug( OSRF_LOG_MARK,  "App Session [%s] [%s] resetting remote id to %s",
467                         session->remote_service, session->session_id, session->orig_remote_id );
468
469         osrf_app_session_set_remote( session, session->orig_remote_id );
470 }
471
472 void osrf_app_session_set_remote( osrfAppSession* session, const char* remote_id ) {
473         if(session == NULL)
474                 return;
475
476         if( session->remote_id ) {
477                 if( strlen(session->remote_id) >= strlen(remote_id) ) {
478                         // There's enough room; just copy it
479                         strcpy(session->remote_id, remote_id);
480                 } else {
481                         free(session->remote_id );
482                         session->remote_id = strdup( remote_id );
483                 }
484         } else
485                 session->remote_id = strdup( remote_id );
486 }
487
488 /** pushes the given message into the result list of the app_request
489   with the given request_id */
490 int osrf_app_session_push_queue( 
491                 osrfAppSession* session, osrfMessage* msg ){
492         if(session == NULL || msg == NULL) return 0;
493
494         osrfAppRequest* req = OSRF_LIST_GET_INDEX( session->request_queue, msg->thread_trace );
495         if(req == NULL) return 0;
496         _osrf_app_request_push_queue( req, msg );
497
498         return 0;
499 }
500
501 /** Attempts to connect to the remote service */
502 int osrfAppSessionConnect( osrfAppSession* session ) {
503         
504         if(session == NULL)
505                 return 0;
506
507         if(session->state == OSRF_SESSION_CONNECTED) {
508                 return 1;
509         }
510
511         int timeout = 5; /* XXX CONFIG VALUE */
512
513         osrfLogDebug( OSRF_LOG_MARK,  "AppSession connecting to %s", session->remote_id );
514
515         /* defaulting to protocol 1 for now */
516         osrfMessage* con_msg = osrf_message_init( CONNECT, session->thread_trace, 1 );
517         osrf_app_session_reset_remote( session );
518         session->state = OSRF_SESSION_CONNECTING;
519         int ret = _osrf_app_session_send( session, con_msg );
520         osrfMessageFree(con_msg);
521         if(ret) return 0;
522
523         time_t start = time(NULL);      
524         time_t remaining = (time_t) timeout;
525
526         while( session->state != OSRF_SESSION_CONNECTED && remaining >= 0 ) {
527                 osrf_app_session_queue_wait( session, remaining, NULL );
528         if(session->transport_error) {
529             osrfLogError(OSRF_LOG_MARK, "cannot communicate with %s", session->remote_service);
530             return 0;
531         }
532                 remaining -= (int) (time(NULL) - start);
533         }
534
535         if(session->state == OSRF_SESSION_CONNECTED)
536                 osrfLogDebug( OSRF_LOG_MARK, " * Connected Successfully to %s", session->remote_service );
537
538         if(session->state != OSRF_SESSION_CONNECTED)
539                 return 0;
540
541         return 1;
542 }
543
544
545
546 /** Disconnects from the remote service */
547 int osrf_app_session_disconnect( osrfAppSession* session){
548         if(session == NULL)
549                 return 1;
550
551         if(session->state == OSRF_SESSION_DISCONNECTED)
552                 return 1;
553
554         if(session->stateless && session->state != OSRF_SESSION_CONNECTED) {
555                 osrfLogDebug( OSRF_LOG_MARK,  
556                                 "Exiting disconnect on stateless session %s", 
557                                 session->session_id);
558                 return 1;
559         }
560
561         osrfLogDebug(OSRF_LOG_MARK,  "AppSession disconnecting from %s", session->remote_id );
562
563         osrfMessage* dis_msg = osrf_message_init( DISCONNECT, session->thread_trace, 1 );
564         _osrf_app_session_send( session, dis_msg );
565         session->state = OSRF_SESSION_DISCONNECTED;
566
567         osrfMessageFree( dis_msg );
568         osrf_app_session_reset_remote( session );
569         return 1;
570 }
571
572 int osrf_app_session_request_resend( osrfAppSession* session, int req_id ) {
573         osrfAppRequest* req = OSRF_LIST_GET_INDEX( session->request_queue, req_id );
574         return _osrf_app_request_resend( req );
575 }
576
577
578 static int osrfAppSessionSendBatch( osrfAppSession* session, osrfMessage* msgs[], int size ) {
579
580         if( !(session && msgs && size > 0) ) return 0;
581         int retval = 0;
582
583         osrfMessage* msg = msgs[0];
584
585         if(msg) {
586
587                 osrf_app_session_queue_wait( session, 0, NULL );
588
589                 if(session->state != OSRF_SESSION_CONNECTED)  {
590
591                         if(session->stateless) { /* stateless session always send to the root listener */
592                                 osrf_app_session_reset_remote(session);
593
594                         } else { 
595
596                                 /* do an auto-connect if necessary */
597                                 if( ! session->stateless &&
598                                         (msg->m_type != CONNECT) && 
599                                         (msg->m_type != DISCONNECT) &&
600                                         (session->state != OSRF_SESSION_CONNECTED) ) {
601
602                                         if(!osrf_app_session_connect( session )) 
603                                                 return 0;
604                                 }
605                         }
606                 }
607         }
608
609         char* string = osrfMessageSerializeBatch(msgs, size);
610
611         if( string ) {
612
613                 transport_message* t_msg = message_init( 
614                                 string, "", session->session_id, session->remote_id, NULL );
615       message_set_osrf_xid( t_msg, osrfLogGetXid() );
616
617                 retval = client_send_message( session->transport_handle, t_msg );
618
619                 if( retval ) osrfLogError(OSRF_LOG_MARK, "client_send_message failed");
620
621                 osrfLogInfo(OSRF_LOG_MARK, "[%s] sent %d bytes of data to %s",
622                         session->remote_service, strlen(string), t_msg->recipient );
623
624                 osrfLogDebug(OSRF_LOG_MARK, "Sent: %s", string );
625
626                 free(string);
627                 message_free( t_msg );
628         }
629
630         return retval; 
631 }
632
633
634
635 static int _osrf_app_session_send( osrfAppSession* session, osrfMessage* msg ){
636         if( !(session && msg) ) return 0;
637         osrfMessage* a[1];
638         a[0] = msg;
639         return osrfAppSessionSendBatch( session, a, 1 );
640 }
641
642
643
644
645 /**  Waits up to 'timeout' seconds for some data to arrive.
646   * Any data that arrives will be processed according to its
647   * payload and message type.  This method will return after
648   * any data has arrived.
649   */
650 int osrf_app_session_queue_wait( osrfAppSession* session, int timeout, int* recvd ){
651         if(session == NULL) return 0;
652         osrfLogDebug(OSRF_LOG_MARK,  "AppSession in queue_wait with timeout %d", timeout );
653         return osrf_stack_entry_point(session->transport_handle, timeout, recvd);
654 }
655
656 /** Disconnects (if client) and removes the given session from the global session cache 
657   * ! This frees all attached app_requests !
658   */
659 void osrfAppSessionFree( osrfAppSession* session ){
660         if(session == NULL) return;
661
662         /* Disconnect */
663         
664         osrfLogDebug(OSRF_LOG_MARK,  "AppSession [%s] [%s] destroying self and deleting requests", 
665                         session->remote_service, session->session_id );
666         if(session->type == OSRF_SESSION_CLIENT 
667                         && session->state != OSRF_SESSION_DISCONNECTED ) { /* disconnect if we're a client */
668                 osrfMessage* dis_msg = osrf_message_init( DISCONNECT, session->thread_trace, 1 );
669                 _osrf_app_session_send( session, dis_msg ); 
670                 osrfMessageFree(dis_msg);
671         }
672
673         /* Remove self from the global session cache */
674         
675         osrfHashRemove( osrfAppSessionCache, session->session_id );
676         
677         /* Free the memory */
678         
679         if( session->userDataFree && session->userData )
680                 session->userDataFree(session->userData);
681         
682         if(session->session_locale)
683                 free(session->session_locale);
684
685         free(session->remote_id);
686         free(session->orig_remote_id);
687         free(session->session_id);
688         free(session->remote_service);
689         osrfListFree(session->request_queue);
690         free(session);
691 }
692
693 osrfMessage* osrfAppSessionRequestRecv(
694                 osrfAppSession* session, int req_id, int timeout ) {
695         if(req_id < 0 || session == NULL)
696                 return NULL;
697         osrfAppRequest* req = OSRF_LIST_GET_INDEX( session->request_queue, req_id );
698         return _osrf_app_request_recv( req, timeout );
699 }
700
701
702
703 int osrfAppRequestRespond( osrfAppSession* ses, int requestId, const jsonObject* data ) {
704         if(!ses || ! data ) return -1;
705
706         osrfMessage* msg = osrf_message_init( RESULT, requestId, 1 );
707         osrf_message_set_status_info( msg, NULL, "OK", OSRF_STATUS_OK );
708         char* json = jsonObjectToJSON( data );
709
710         osrf_message_set_result_content( msg, json );
711         _osrf_app_session_send( ses, msg ); 
712
713         free(json);
714         osrfMessageFree( msg );
715
716         return 0;
717 }
718
719
720 int osrfAppRequestRespondComplete( 
721                 osrfAppSession* ses, int requestId, const jsonObject* data ) {
722
723         osrfMessage* status = osrf_message_init( STATUS, requestId, 1);
724         osrf_message_set_status_info( status, "osrfConnectStatus", "Request Complete", OSRF_STATUS_COMPLETE );
725         
726         if (data) {
727                 osrfMessage* payload = osrf_message_init( RESULT, requestId, 1 );
728                 osrf_message_set_status_info( payload, NULL, "OK", OSRF_STATUS_OK );
729
730                 char* json = jsonObjectToJSON( data );
731                 osrf_message_set_result_content( payload, json );
732                 free(json);
733
734                 osrfMessage* ms[2];
735                 ms[0] = payload;
736                 ms[1] = status;
737
738                 osrfAppSessionSendBatch( ses, ms, 2 );
739
740                 osrfMessageFree( payload );
741         } else {
742                 osrfAppSessionSendBatch( ses, &status, 1 );
743         }
744
745         osrfMessageFree( status );
746
747         return 0;
748 }
749
750 int osrfAppSessionStatus( osrfAppSession* ses, int type,
751                 const char* name, int reqId, const char* message ) {
752
753         if(ses) {
754                 osrfMessage* msg = osrf_message_init( STATUS, reqId, 1);
755                 osrf_message_set_status_info( msg, name, message, type );
756                 _osrf_app_session_send( ses, msg ); 
757                 osrfMessageFree( msg );
758                 return 0;
759         }
760         return -1;
761 }
762
763
764
765
766
767