015985d8ffe3412e7ac405a4eced0c7d36207cde
[OpenSRF.git] / src / perlmods / OpenSRF / AppSession.pm
1 package OpenSRF::AppSession;
2 use OpenSRF::DOM;
3 #use OpenSRF::DOM::Element::userAuth;
4 use OpenSRF::DomainObject::oilsMessage;
5 use OpenSRF::DomainObject::oilsMethod;
6 use OpenSRF::DomainObject::oilsResponse qw/:status/;
7 use OpenSRF::Transport::PeerHandle;
8 use OpenSRF::Utils::Logger qw(:level);
9 use OpenSRF::Utils::SettingsClient;
10 use OpenSRF::Utils::Config;
11 use OpenSRF::EX;
12 use OpenSRF;
13 use Exporter;
14 use base qw/Exporter OpenSRF/;
15 use Time::HiRes qw( time usleep );
16 use warnings;
17 use strict;
18
19 our @EXPORT_OK = qw/CONNECTING INIT_CONNECTED CONNECTED DISCONNECTED CLIENT SERVER/;
20 our %EXPORT_TAGS = ( state => [ qw/CONNECTING INIT_CONNECTED CONNECTED DISCONNECTED/ ],
21                  endpoint => [ qw/CLIENT SERVER/ ],
22 );
23
24 my $logger = "OpenSRF::Utils::Logger";
25
26 our %_CACHE;
27 our @_RESEND_QUEUE;
28
29 sub CONNECTING { return 3 };
30 sub INIT_CONNECTED { return 4 };
31 sub CONNECTED { return 1 };
32 sub DISCONNECTED { return 2 };
33
34 sub CLIENT { return 2 };
35 sub SERVER { return 1 };
36
37 sub find {
38         return undef unless (defined $_[1]);
39         return $_CACHE{$_[1]} if (exists($_CACHE{$_[1]}));
40 }
41
42 sub transport_connected {
43         my $self = shift;
44         if( ! exists $self->{peer_handle} || ! $self->{peer_handle} ) {
45                 return 0;
46         }
47         return $self->{peer_handle}->tcp_connected();
48 }
49
50 sub connected {
51         my $self = shift;
52         return $self->state == CONNECTED;
53 }
54 # ----------------------------------------------------------------------------
55 # Clears the transport buffers
56 # call this if you are not through with the sesssion, but you want 
57 # to have a clean slate.  You shouldn't have to call this if
58 # you are correctly 'recv'ing all of the data from a request.
59 # however, if you don't want all of the data, this will
60 # slough off any excess
61 #  * * Note: This will delete data for all sessions using this transport
62 # handle.  For example, all client sessions use the same handle.
63 # ----------------------------------------------------------------------------
64 sub buffer_reset {
65
66         my $self = shift;
67         if( ! exists $self->{peer_handle} || ! $self->{peer_handle} ) { 
68                 return 0;
69         }
70         $self->{peer_handle}->buffer_reset();
71 }
72
73
74 # when any incoming data is received, this method is called.
75 sub server_build {
76         my $class = shift;
77         $class = ref($class) || $class;
78
79         my $sess_id = shift;
80         my $remote_id = shift;
81         my $service = shift;
82
83         warn "Missing args to server_build():\n" .
84                 "sess_id: $sess_id, remote_id: $remote_id, service: $service\n" 
85                 unless ($sess_id and $remote_id and $service);
86
87         return undef unless ($sess_id and $remote_id and $service);
88
89         if ( my $thingy = $class->find($sess_id) ) {
90                 $thingy->remote_id( $remote_id );
91                 return $thingy;
92         }
93
94         if( $service eq "client" ) {
95                 #throw OpenSRF::EX::PANIC ("Attempting to build a client session as a server" .
96                 #       " Session ID [$sess_id], remote_id [$remote_id]");
97
98                 warn "Attempting to build a client session as ".
99                                 "a server Session ID [$sess_id], remote_id [$remote_id]";
100
101                 $logger->debug("Attempting to build a client session as ".
102                                 "a server Session ID [$sess_id], remote_id [$remote_id]", ERROR );
103
104                 return undef;
105         }
106
107         my $config_client = OpenSRF::Utils::SettingsClient->new();
108         my $stateless = $config_client->config_value("apps", $service, "stateless");
109
110         #my $max_requests = $conf->$service->max_requests;
111         my $max_requests        = $config_client->config_value("apps",$service,"max_requests");
112         $logger->debug( "Max Requests for $service is $max_requests", INTERNAL ) if (defined $max_requests);
113
114         $logger->transport( "AppSession creating new session: $sess_id", INTERNAL );
115
116         my $self = bless { recv_queue  => [],
117                            request_queue  => [],
118                            requests  => 0,
119                            session_data  => {},
120                            callbacks  => {},
121                            endpoint    => SERVER,
122                            state       => CONNECTING, 
123                            session_id  => $sess_id,
124                            remote_id    => $remote_id,
125                                 peer_handle => OpenSRF::Transport::PeerHandle->retrieve($service),
126                                 max_requests => $max_requests,
127                                 session_threadTrace => 0,
128                                 service => $service,
129                                 stateless => $stateless,
130                          } => $class;
131
132         return $_CACHE{$sess_id} = $self;
133 }
134
135 sub session_data {
136         my $self = shift;
137         my ($name, $datum) = @_;
138
139         $self->{session_data}->{$name} = $datum if (defined $datum);
140         return $self->{session_data}->{$name};
141 }
142
143 sub service { return shift()->{service}; }
144
145 sub continue_request {
146         my $self = shift;
147         $self->{'requests'}++;
148         return 1 if (!$self->{'max_requests'});
149         return $self->{'requests'} <= $self->{'max_requests'} ? 1 : 0;
150 }
151
152 sub last_sent_payload {
153         my( $self, $payload ) = @_;
154         if( $payload ) {
155                 return $self->{'last_sent_payload'} = $payload;
156         }
157         return $self->{'last_sent_payload'};
158 }
159
160 sub last_sent_type {
161         my( $self, $type ) = @_;
162         if( $type ) {
163                 return $self->{'last_sent_type'} = $type;
164         }
165         return $self->{'last_sent_type'};
166 }
167
168 sub get_app_targets {
169         my $app = shift;
170
171         my $conf = OpenSRF::Utils::Config->current;
172         my $router_name = $conf->bootstrap->router_name || 'router';
173         my $routers = $conf->bootstrap->domains;
174
175         unless($router_name and $routers) {
176                 throw OpenSRF::EX::Config 
177                         ("Missing router config information 'router_name' and 'routers'");
178         }
179
180         my @targets;
181         for my $router (@$routers) {
182                 push @targets, "$router_name\@$router/$app";
183         }
184
185         return @targets;
186 }
187
188 sub stateless {
189         my $self = shift;
190         my $state = shift;
191         $self->{stateless} = $state if (defined $state);
192         return $self->{stateless};
193 }
194
195 # When we're a client and we want to connect to a remote service
196 # create( $app, username => $user, secret => $passwd );
197 #    OR
198 # create( $app, sysname => $user, secret => $shared_secret );
199 sub create {
200         my $class = shift;
201         $class = ref($class) || $class;
202
203         my $app = shift;
204         my $api_level = shift;
205         my $quiet = shift;
206
207         $api_level = 1 if (!defined($api_level));
208                                 
209         $logger->debug( "AppSession creating new client session for $app", DEBUG );
210
211         my $stateless = 0;
212         my $c = OpenSRF::Utils::SettingsClient->new();
213         # we can get an infinite loop if we're grabbing the settings and we
214         # need the settings to grab the settings...
215         if($app ne "opensrf.settings" || $c->has_config()) { 
216                 $stateless = $c->config_value("apps", $app, "stateless");
217         }
218
219         my $sess_id = time . rand( $$ );
220         while ( $class->find($sess_id) ) {
221                 $sess_id = time . rand( $$ );
222         }
223
224         
225         my ($r_id) = get_app_targets($app);
226
227         my $peer_handle = OpenSRF::Transport::PeerHandle->retrieve("client"); 
228         if( ! $peer_handle ) {
229                 $peer_handle = OpenSRF::Transport::PeerHandle->retrieve("system_client");
230         }
231
232         my $self = bless { app_name    => $app,
233                                 #client_auth => $auth,
234                            #recv_queue  => [],
235                            request_queue  => [],
236                            endpoint    => CLIENT,
237                            state       => DISCONNECTED,#since we're init'ing
238                            session_id  => $sess_id,
239                            remote_id   => $r_id,
240                            raise_error   => $quiet ? 0 : 1,
241                            api_level   => $api_level,
242                            orig_remote_id   => $r_id,
243                                 peer_handle => $peer_handle,
244                                 session_threadTrace => 0,
245                                 stateless               => $stateless,
246                          } => $class;
247
248         $logger->debug( "Created new client session $app : $sess_id" );
249
250         return $_CACHE{$sess_id} = $self;
251 }
252
253 sub raise_remote_errors {
254         my $self = shift;
255         my $err = shift;
256         $self->{raise_error} = $err if (defined $err);
257         return $self->{raise_error};
258 }
259
260 sub api_level {
261         return shift()->{api_level};
262 }
263
264 sub app {
265         return shift()->{app_name};
266 }
267
268 sub reset {
269         my $self = shift;
270         $self->remote_id($$self{orig_remote_id});
271 }
272
273 # 'connect' can be used as a constructor if called as a class method,
274 # or used to connect a session that has disconnectd if called against
275 # an existing session that seems to be disconnected, or was just built
276 # using 'create' above.
277
278 # connect( $app, username => $user, secret => $passwd );
279 #    OR
280 # connect( $app, sysname => $user, secret => $shared_secret );
281
282 # --- Returns undef if the connect attempt times out.
283 # --- Returns the OpenSRF::EX object if one is returned by the server
284 # --- Returns self if connected
285 sub connect {
286         my $self = shift;
287         my $class = ref($self) || $self;
288
289
290         if ( ref( $self ) and  $self->state && $self->state == CONNECTED  ) {
291                 $logger->transport("AppSession already connected", DEBUG );
292         } else {
293                 $logger->transport("AppSession not connected, connecting..", DEBUG );
294         }
295         return $self if ( ref( $self ) and  $self->state && $self->state == CONNECTED  );
296
297
298         my $app = shift;
299         my $api_level = shift;
300         $api_level = 1 unless (defined $api_level);
301
302         $self = $class->create($app, @_) if (!ref($self));
303
304         return undef unless ($self);
305
306         $self->{api_level} = $api_level;
307
308         $self->reset;
309         $self->state(CONNECTING);
310         $self->send('CONNECT', "");
311
312
313         # if we want to connect to settings, we may not have 
314         # any data for the settings client to work with...
315         # just using a default for now XXX
316
317         my $time_remaining = 5;
318         
319 =head blah
320         my $client = OpenSRF::Utils::SettingsClient->new();
321         my $trans = $client->config_value("client_connection","transport_host");
322
323         if(!ref($trans)) {
324                 $time_remaining = $trans->{connect_timeout};
325         } else {
326                 # XXX for now, just use the first
327                 $time_remaining = $trans->[0]->{connect_timeout};
328         }
329 =cut
330
331         while ( $self->state != CONNECTED  and $time_remaining > 0 ) {
332                 my $starttime = time;
333                 $self->queue_wait($time_remaining);
334                 my $endtime = time;
335                 $time_remaining -= ($endtime - $starttime);
336         }
337
338         return undef unless($self->state == CONNECTED);
339
340         $self->stateless(0);
341
342         return $self;
343 }
344
345 sub finish {
346         my $self = shift;
347         if( ! $self->session_id ) {
348                 return 0;
349         }
350 }
351
352 sub unregister_callback {
353         my $self = shift;
354         my $type = shift;
355         my $cb = shift;
356         if (exists $self->{callbacks}{$type}) {
357                 delete $self->{callbacks}{$type}{$cb};
358                 return $cb;
359         }
360         return undef;
361 }
362
363 sub register_callback {
364         my $self = shift;
365         my $type = shift;
366         my $cb = shift;
367         my $cb_key = "$cb";
368         $self->{callbacks}{$type}{$cb_key} = $cb;
369         return $cb_key;
370 }
371
372 sub kill_me {
373         my $self = shift;
374         if( ! $self->session_id ) { return 0; }
375
376         # run each 'death' callback;
377         if (exists $self->{callbacks}{death}) {
378                 for my $sub (values %{$self->{callbacks}{death}}) {
379                         $sub->($self);
380                 }
381         }
382
383         $self->disconnect;
384         $logger->transport( "AppSession killing self: " . $self->session_id(), DEBUG );
385         delete $_CACHE{$self->session_id};
386         delete($$self{$_}) for (keys %$self);
387 }
388
389 sub disconnect {
390         my $self = shift;
391
392         # run each 'disconnect' callback;
393         if (exists $self->{callbacks}{disconnect}) {
394                 for my $sub (values %{$self->{callbacks}{disconnect}}) {
395                         $sub->($self);
396                 }
397         }
398
399         if ( !$self->stateless and $self->state != DISCONNECTED ) {
400                 $self->send('DISCONNECT', "") if ($self->endpoint == CLIENT);
401                 $self->state( DISCONNECTED ); 
402         }
403
404         $self->reset;
405 }
406
407 sub request {
408         my $self = shift;
409         my $meth = shift;
410         return unless $self;
411
412         my $method;
413         if (!ref $meth) {
414                 $method = new OpenSRF::DomainObject::oilsMethod ( method => $meth );
415         } else {
416                 $method = $meth;
417         }
418         
419         $method->params( @_ );
420
421         $self->send('REQUEST',$method);
422 }
423
424 sub full_request {
425         my $self = shift;
426         my $meth = shift;
427
428         my $method;
429         if (!ref $meth) {
430                 $method = new OpenSRF::DomainObject::oilsMethod ( method => $meth );
431         } else {
432                 $method = $meth;
433         }
434         
435         $method->params( @_ );
436
437         $self->send(CONNECT => '', REQUEST => $method, DISCONNECT => '');
438 }
439
440 sub send {
441         my $self = shift;
442         my @payload_list = @_; # this is a Domain Object
443
444         return unless ($self and $self->{peer_handle});
445
446         $logger->debug( "In send", INTERNAL );
447         
448         my $tT;
449
450         if( @payload_list % 2 ) { $tT = pop @payload_list; }
451
452         if( ! @payload_list ) {
453                 $logger->debug( "payload_list param is incomplete in AppSession::send()", ERROR );
454                 return undef; 
455         }
456
457         my @doc = ();
458
459         my $disconnect = 0;
460         my $connecting = 0;
461
462         while( @payload_list ) {
463
464                 my ($msg_type, $payload) = ( shift(@payload_list), shift(@payload_list) ); 
465
466                 if ($msg_type eq 'DISCONNECT' ) {
467                         $disconnect++;
468                         if( $self->state == DISCONNECTED && !$connecting) {
469                                 next;
470                         }
471                 }
472
473                 if( $msg_type eq "CONNECT" ) { 
474                         $connecting++; 
475                 }
476
477                 my $msg = OpenSRF::DomainObject::oilsMessage->new();
478                 $msg->type($msg_type);
479         
480                 no warnings;
481                 $msg->threadTrace( $tT || int($self->session_threadTrace) || int($self->last_threadTrace) );
482                 use warnings;
483         
484                 if ($msg->type eq 'REQUEST') {
485                         if ( !defined($tT) || $self->last_threadTrace != $tT ) {
486                                 $msg->update_threadTrace;
487                                 $self->session_threadTrace( $msg->threadTrace );
488                                 $tT = $self->session_threadTrace;
489                                 OpenSRF::AppRequest->new($self, $payload);
490                         }
491                 }
492         
493                 $msg->api_level($self->api_level);
494                 $msg->payload($payload) if $payload;
495         
496                 push @doc, $msg;
497
498         
499                 $logger->debug( "AppSession sending ".$msg->type." to ".$self->remote_id.
500                         " with threadTrace [".$msg->threadTrace."]" );
501
502         }
503         
504         if ($self->endpoint == CLIENT and ! $disconnect) {
505                 $self->queue_wait(0);
506
507
508                 if($self->stateless && $self->state != CONNECTED) {
509                         $self->reset;
510                         $logger->debug("AppSession is stateless in send", INTERNAL );
511                 }
512
513                 if( !$self->stateless and $self->state != CONNECTED ) {
514
515                         $logger->debug( "Sending connect before request 1", INTERNAL );
516
517                         unless (($self->state == CONNECTING && $connecting )) {
518                                 $logger->debug( "Sending connect before request 2", INTERNAL );
519                                 my $v = $self->connect();
520                                 if( ! $v ) {
521                                         $logger->debug( "Unable to connect to remote service in AppSession::send()", ERROR );
522                                         return undef;
523                                 }
524                                 if( ref($v) and $v->can("class") and $v->class->isa( "OpenSRF::EX" ) ) {
525                                         return $v;
526                                 }
527                         }
528                 }
529
530         } 
531         $logger->debug( "AppSession sending doc: " . JSON->perl2JSON(\@doc), INTERNAL );
532
533
534         $self->{peer_handle}->send( 
535                                         to     => $self->remote_id,
536                                    thread => $self->session_id,
537                                    body   => JSON->perl2JSON(\@doc) );
538
539         if( $disconnect) {
540                 $self->state( DISCONNECTED );
541         }
542
543         my $req = $self->app_request( $tT );
544         $req->{_start} = time;
545         return $req
546 }
547
548 sub app_request {
549         my $self = shift;
550         my $tT = shift;
551         
552         return undef unless (defined $tT);
553         my ($req) = grep { $_->threadTrace == $tT } @{ $self->{request_queue} };
554
555         return $req;
556 }
557
558 sub remove_app_request {
559         my $self = shift;
560         my $req = shift;
561         
562         my @list = grep { $_->threadTrace != $req->threadTrace } @{ $self->{request_queue} };
563
564         $self->{request_queue} = \@list;
565 }
566
567 sub endpoint {
568         return $_[0]->{endpoint};
569 }
570
571
572 sub session_id {
573         my $self = shift;
574         return $self->{session_id};
575 }
576
577 sub push_queue {
578         my $self = shift;
579         my $resp = shift;
580         my $req = $self->app_request($resp->[1]);
581         return $req->push_queue( $resp->[0] ) if ($req);
582         push @{ $self->{recv_queue} }, $resp->[0];
583 }
584
585 sub last_threadTrace {
586         my $self = shift;
587         my $new_last_threadTrace = shift;
588
589         my $old_last_threadTrace = $self->{last_threadTrace};
590         if (defined $new_last_threadTrace) {
591                 $self->{last_threadTrace} = $new_last_threadTrace;
592                 return $new_last_threadTrace unless ($old_last_threadTrace);
593         }
594
595         return $old_last_threadTrace;
596 }
597
598 sub session_threadTrace {
599         my $self = shift;
600         my $new_last_threadTrace = shift;
601
602         my $old_last_threadTrace = $self->{session_threadTrace};
603         if (defined $new_last_threadTrace) {
604                 $self->{session_threadTrace} = $new_last_threadTrace;
605                 return $new_last_threadTrace unless ($old_last_threadTrace);
606         }
607
608         return $old_last_threadTrace;
609 }
610
611 sub last_message_type {
612         my $self = shift;
613         my $new_last_message_type = shift;
614
615         my $old_last_message_type = $self->{last_message_type};
616         if (defined $new_last_message_type) {
617                 $self->{last_message_type} = $new_last_message_type;
618                 return $new_last_message_type unless ($old_last_message_type);
619         }
620
621         return $old_last_message_type;
622 }
623
624 sub last_message_api_level {
625         my $self = shift;
626         my $new_last_message_api_level = shift;
627
628         my $old_last_message_api_level = $self->{last_message_api_level};
629         if (defined $new_last_message_api_level) {
630                 $self->{last_message_api_level} = $new_last_message_api_level;
631                 return $new_last_message_api_level unless ($old_last_message_api_level);
632         }
633
634         return $old_last_message_api_level;
635 }
636
637 sub remote_id {
638         my $self = shift;
639         my $new_remote_id = shift;
640
641         my $old_remote_id = $self->{remote_id};
642         if (defined $new_remote_id) {
643                 $self->{remote_id} = $new_remote_id;
644                 return $new_remote_id unless ($old_remote_id);
645         }
646
647         return $old_remote_id;
648 }
649
650 sub client_auth {
651         return undef;
652         my $self = shift;
653         my $new_ua = shift;
654
655         my $old_ua = $self->{client_auth};
656         if (defined $new_ua) {
657                 $self->{client_auth} = $new_ua;
658                 return $new_ua unless ($old_ua);
659         }
660
661         return $old_ua->cloneNode(1);
662 }
663
664 sub state {
665         my $self = shift;
666         my $new_state = shift;
667
668         my $old_state = $self->{state};
669         if (defined $new_state) {
670                 $self->{state} = $new_state;
671                 return $new_state unless ($old_state);
672         }
673
674         return $old_state;
675 }
676
677 sub DESTROY {
678         my $self = shift;
679         delete $$self{$_} for keys %$self;
680         return undef;
681 }
682
683 sub recv {
684         my $self = shift;
685         my @proto_args = @_;
686         my %args;
687
688         if ( @proto_args ) {
689                 if ( !(@proto_args % 2) ) {
690                         %args = @proto_args;
691                 } elsif (@proto_args == 1) {
692                         %args = ( timeout => @proto_args );
693                 }
694         }
695
696         #$logger->debug( ref($self). " recv_queue before wait: " . $self->_print_queue(), INTERNAL );
697
698         if( exists( $args{timeout} ) ) {
699                 $args{timeout} = int($args{timeout});
700                 $self->{recv_timeout} = $args{timeout};
701         }
702
703         #$args{timeout} = 0 if ($self->complete);
704
705         if(defined($args{timeout})) {
706                 $logger->debug( ref($self) ."->recv with timeout " . $args{timeout}, INTERNAL );
707         }
708
709         my $avail = @{ $self->{recv_queue} };
710         $self->{remaining_recv_timeout} = $self->{recv_timeout};
711
712         if (!$args{count}) {
713                 if (wantarray) {
714                         $args{count} = $avail;
715                 } else {
716                         $args{count} = 1;
717                 }
718         }
719
720         while ( $self->{remaining_recv_timeout} > 0 and $avail < $args{count} ) {
721                         last if $self->complete;
722                         my $starttime = time;
723                         $self->queue_wait($self->{remaining_recv_timeout});
724                         my $endtime = time;
725                         if ($self->{timeout_reset}) {
726                                 $self->{timeout_reset} = 0;
727                         } else {
728                                 $self->{remaining_recv_timeout} -= ($endtime - $starttime)
729                         }
730                         $avail = @{ $self->{recv_queue} };
731         }
732
733
734         my @list;
735         while ( my $msg = shift @{ $self->{recv_queue} } ) {
736                 push @list, $msg;
737                 last if (scalar(@list) >= $args{count});
738         }
739
740         $logger->debug( "Number of matched responses: " . @list, DEBUG );
741         $self->queue_wait(0); # check for statuses
742         
743         return $list[0] unless (wantarray);
744         return @list;
745 }
746
747 sub push_resend {
748         my $self = shift;
749         push @OpenSRF::AppSession::_RESEND_QUEUE, @_;
750 }
751
752 sub flush_resend {
753         my $self = shift;
754         $logger->debug( "Resending..." . @_RESEND_QUEUE, INTERNAL );
755         while ( my $req = shift @OpenSRF::AppSession::_RESEND_QUEUE ) {
756                 $req->resend unless $req->complete;
757         }
758 }
759
760
761 sub queue_wait {
762         my $self = shift;
763         if( ! $self->{peer_handle} ) { return 0; }
764         my $timeout = shift || 0;
765         $logger->debug( "Calling queue_wait($timeout)" , INTERNAL );
766         my $o = $self->{peer_handle}->process($timeout);
767         $self->flush_resend;
768         return $o;
769 }
770
771 sub _print_queue {
772         my( $self ) = @_;
773         my $string = "";
774         foreach my $msg ( @{$self->{recv_queue}} ) {
775                 $string = $string . $msg->toString(1) . "\n";
776         }
777         return $string;
778 }
779
780 sub status {
781         my $self = shift;
782         return unless $self;
783         $self->send( 'STATUS', @_ );
784 }
785
786 sub reset_request_timeout {
787         my $self = shift;
788         my $tt = shift;
789         my $req = $self->app_request($tt);
790         $req->{remaining_recv_timeout} = $req->{recv_timeout};
791         $req->{timout_reset} = 1;
792 }
793
794 #-------------------------------------------------------------------------------
795
796 package OpenSRF::AppRequest;
797 use base qw/OpenSRF::AppSession/;
798 use OpenSRF::Utils::Logger qw/:level/;
799 use OpenSRF::DomainObject::oilsResponse qw/:status/;
800 use Time::HiRes qw/time usleep/;
801
802 sub new {
803         my $class = shift;
804         $class = ref($class) || $class;
805
806         my $session = shift;
807         my $threadTrace = $session->session_threadTrace || $session->last_threadTrace;
808         my $payload = shift;
809         
810         my $self = {    session                 => $session,
811                         threadTrace             => $threadTrace,
812                         payload                 => $payload,
813                         complete                => 0,
814                         timeout_reset           => 0,
815                         recv_timeout            => 30,
816                         remaining_recv_timeout  => 30,
817                         recv_queue              => [],
818         };
819
820         bless $self => $class;
821
822         push @{ $self->session->{request_queue} }, $self;
823
824         return $self;
825 }
826
827 sub recv_timeout {
828         my $self = shift;
829         my $timeout = shift;
830         if (defined $timeout) {
831                 $self->{recv_timeout} = $timeout;
832                 $self->{remaining_recv_timeout} = $timeout;
833         }
834         return $self->{recv_timeout};
835 }
836
837 sub queue_size {
838         my $size = @{$_[0]->{recv_queue}};
839         return $size;
840 }
841         
842 sub send {
843         my $self = shift;
844         return unless ($self and $self->session and !$self->complete);
845         $self->session->send(@_);
846 }
847
848 sub finish {
849         my $self = shift;
850         return unless $self->session;
851         $self->session->remove_app_request($self);
852         delete($$self{$_}) for (keys %$self);
853 }
854
855 sub session {
856         return shift()->{session};
857 }
858
859 sub complete {
860         my $self = shift;
861         my $complete = shift;
862         return $self->{complete} if ($self->{complete});
863         if (defined $complete) {
864                 $self->{complete} = $complete;
865                 $self->{_duration} = time - $self->{_start} if ($self->{complete});
866         } else {
867                 $self->session->queue_wait(0);
868         }
869         return $self->{complete};
870 }
871
872 sub duration {
873         my $self = shift;
874         $self->wait_complete;
875         return $self->{_duration};
876 }
877
878 sub wait_complete {
879         my $self = shift;
880         my $timeout = shift || 10;
881         my $time_remaining = $timeout;
882
883         while ( ! $self->complete  and $time_remaining > 0 ) {
884                 my $starttime = time;
885                 $self->queue_wait($time_remaining);
886                 my $endtime = time;
887                 $time_remaining -= ($endtime - $starttime);
888         }
889
890         return $self->complete;
891 }
892
893 sub threadTrace {
894         return shift()->{threadTrace};
895 }
896
897 sub push_queue {
898         my $self = shift;
899         my $resp = shift;
900         if( !$resp ) { return 0; }
901         if( UNIVERSAL::isa($resp, "Error")) {
902                 $self->{failed} = $resp;
903                 $self->complete(1);
904                 #return; eventually...
905         }
906         push @{ $self->{recv_queue} }, $resp;
907 }
908
909 sub failed {
910         my $self = shift;
911         return $self->{failed};
912 }
913
914 sub queue_wait {
915         my $self = shift;
916         return $self->session->queue_wait(@_)
917 }
918
919 sub payload { return shift()->{payload}; }
920
921 sub resend {
922         my $self = shift;
923         return unless ($self and $self->session and !$self->complete);
924         OpenSRF::Utils::Logger->debug( "I'm resending the request for threadTrace ". $self->threadTrace, DEBUG);
925         return $self->session->send('REQUEST', $self->payload, $self->threadTrace );
926 }
927
928 sub status {
929         my $self = shift;
930         my $msg = shift;
931         return unless ($self and $self->session and !$self->complete);
932         $self->session->send( 'STATUS',$msg, $self->threadTrace );
933 }
934
935 sub stream_push {
936         my $self = shift;
937         my $msg = shift;
938         $self->respond( $msg );
939 }
940
941 sub respond {
942         my $self = shift;
943         my $msg = shift;
944         return unless ($self and $self->session and !$self->complete);
945
946         my $response;
947         if (ref($msg) && UNIVERSAL::isa($msg, 'OpenSRF::DomainObject::oilsResult')) {
948                 $response = $msg;
949         } else {
950                 $response = new OpenSRF::DomainObject::oilsResult;
951                 $response->content($msg);
952         }
953
954         $self->session->send('RESULT', $response, $self->threadTrace);
955 }
956
957 sub respond_complete {
958         my $self = shift;
959         my $msg = shift;
960         return unless ($self and $self->session and !$self->complete);
961
962         my $response;
963         if (ref($msg) && UNIVERSAL::isa($msg, 'OpenSRF::DomainObject::oilsResult')) {
964                 $response = $msg;
965         } else {
966                 $response = new OpenSRF::DomainObject::oilsResult;
967                 $response->content($msg);
968         }
969
970         my $stat = OpenSRF::DomainObject::oilsConnectStatus->new(
971                 statusCode => STATUS_COMPLETE(),
972                 status => 'Request Complete' );
973
974
975         $self->session->send( 'RESULT' => $response, 'STATUS' => $stat, $self->threadTrace);
976         $self->complete(1);
977 }
978
979 sub register_death_callback {
980         my $self = shift;
981         my $cb = shift;
982         $self->session->register_callback( death => $cb );
983 }
984
985
986 # utility method.  checks to see of the request failed.
987 # if so, throws an OpenSRF::EX::ERROR. if everything is
988 # ok, it returns the content of the request
989 sub gather {
990         my $self = shift;
991         my $finish = shift;
992         $self->wait_complete;
993         my $resp = $self->recv( timeout => 60 );
994         if( $self->failed() ) { 
995                 throw OpenSRF::EX::ERROR
996                         ($self->failed()->stringify());
997         }
998         if(!$resp) { return undef; }
999         my $content = $resp->content;
1000         if($finish) { $self->finish();}
1001         return $content;
1002 }
1003
1004
1005 package OpenSRF::AppSubrequest;
1006
1007 sub respond {
1008         my $self = shift;
1009         my $resp = shift;
1010         push @{$$self{resp}}, $resp if (defined $resp);
1011 }
1012 sub respond_complete { respond(@_); }
1013
1014 sub new {
1015         my $class = shift;
1016         $class = ref($class) || $class;
1017         return bless({resp => [], @_}, $class);
1018 }
1019
1020 sub responses { @{$_[0]->{resp}} }
1021
1022 sub status {}
1023
1024
1025 1;
1026