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::Config;
   9 use OpenSRF::Utils::Logger qw(:level);
  10 use OpenSRF::EX;
  11 use OpenSRF;
  12 use Exporter;
  13 use base qw/Exporter OpenSRF/;
  14 use Time::HiRes qw( time usleep );
  15 use warnings;
  16 use strict;
  17
  18 our @EXPORT_OK = qw/CONNECTING INIT_CONNECTED CONNECTED DISCONNECTED CLIENT SERVER/;
  19 our %EXPORT_TAGS = ( state => [ qw/CONNECTING INIT_CONNECTED CONNECTED DISCONNECTED/ ],
  20                  endpoint => [ qw/CLIENT SERVER/ ],
  21 );
  22
  23 my $logger = "OpenSRF::Utils::Logger";
  24
  25 our %_CACHE;
  26 our @_CLIENT_CACHE;
  27 our @_RESEND_QUEUE;
  28
  29 sub kill_client_session_cache {
  30         for my $session ( @_CLIENT_CACHE ) {
  31                 $session->kill_me;
  32         }
  33 }
  34
  35 sub CONNECTING { return 3 };
  36 sub INIT_CONNECTED { return 4 };
  37 sub CONNECTED { return 1 };
  38 sub DISCONNECTED { return 2 };
  39
  40 sub CLIENT { return 2 };
  41 sub SERVER { return 1 };
  42
  43 sub find {
  44         return undef unless (defined $_[1]);
  45         return $_CACHE{$_[1]} if (exists($_CACHE{$_[1]}));
  46 }
  47
  48 sub find_client {
  49         my( $self, $app ) = @_;
  50         $logger->debug( "Client Cache contains: " .scalar(@_CLIENT_CACHE), INTERNAL );
  51         my ($client) = grep { $_->[0] eq $app and $_->[1] == 1 } @_CLIENT_CACHE;
  52         $client->[1] = 0;
  53         return $client->[2];
  54 }
  55
  56 sub transport_connected {
  57         my $self = shift;
  58         if( ! exists $self->{peer_handle} || ! $self->{peer_handle} ) {
  59                 return 0;
  60         }
  61         return $self->{peer_handle}->tcp_connected();
  62 }
  63
  64 # ----------------------------------------------------------------------------
  65 # Clears the transport buffers
  66 # call this if you are not through with the sesssion, but you want
  67 # to have a clean slate.  You shouldn't have to call this if
  68 # you are correctly 'recv'ing all of the data from a request.
  69 # however, if you don't want all of the data, this will
  70 # slough off any excess
  71 #  * * Note: This will delete data for all sessions using this transport
  72 # handle.  For example, all client sessions use the same handle.
  73 # ----------------------------------------------------------------------------
  74 sub buffer_reset {
  75
  76         my $self = shift;
  77         if( ! exists $self->{peer_handle} || ! $self->{peer_handle} ) {
  78                 return 0;
  79         }
  80         $self->{peer_handle}->buffer_reset();
  81 }
  82
  83
  84 sub client_cache {
  85         my $self = shift;
  86         push @_CLIENT_CACHE, [ $self->app, 1, $self ];
  87 }
  88
  89 # when any incoming data is received, this method is called.
  90 sub server_build {
  91         my $class = shift;
  92         $class = ref($class) || $class;
  93
  94         my $sess_id = shift;
  95         my $remote_id = shift;
  96         my $service = shift;
  97
  98         return undef unless ($sess_id and $remote_id and $service);
  99
 100         my $conf = OpenSRF::Utils::Config->current;
 101
 102         if( ! $conf ) {
 103                 OpenSRF::EX::Config->throw( "No suitable config found" );
 104         }
 105
 106         if ( my $thingy = $class->find($sess_id) ) {
 107                 $thingy->remote_id( $remote_id );
 108                 $logger->debug( "AppSession returning existing session $sess_id", DEBUG );
 109                 return $thingy;
 110         } else {
 111                 $logger->debug( "AppSession building new server session $sess_id", DEBUG );
 112         }
 113
 114         if( $service eq "client" ) {
 115                 #throw OpenSRF::EX::PANIC ("Attempting to build a client session as a server" .
 116                 #       " Session ID [$sess_id], remote_id [$remote_id]");
 117                 $logger->debug("Attempting to build a client session as ".
 118                                 "a server Session ID [$sess_id], remote_id [$remote_id]", ERROR );
 119         }
 120
 121
 122         my $max_requests = $conf->$service->max_requests;
 123         $logger->debug( "Max Requests for $service is $max_requests", INTERNAL );# if $max_requests;
 124
 125         $logger->transport( "AppSession creating new session: $sess_id", INTERNAL );
 126
 127         my $self = bless { recv_queue  => [],
 128                            request_queue  => [],
 129                            requests  => 0,
 130                            endpoint    => SERVER,
 131                            state       => CONNECTING,
 132                            session_id  => $sess_id,
 133                            remote_id    => $remote_id,
 134                                 peer_handle => OpenSRF::Transport::PeerHandle->retrieve($service),
 135                                 max_requests => $max_requests,
 136                                 session_threadTrace => 0,
 137                                 service => $service,
 138                          } => $class;
 139
 140         return $_CACHE{$sess_id} = $self;
 141 }
 142
 143 sub service { return shift()->{service}; }
 144
 145 sub continue_request {
 146         my $self = shift;
 147         $self->{'requests'}++;
 148         return $self->{'requests'} <= $self->{'max_requests'} ? 1 : 0;
 149 }
 150
 151 sub last_sent_payload {
 152         my( $self, $payload ) = @_;
 153         if( $payload ) {
 154                 return $self->{'last_sent_payload'} = $payload;
 155         }
 156         return $self->{'last_sent_payload'};
 157 }
 158
 159 sub last_sent_type {
 160         my( $self, $type ) = @_;
 161         if( $type ) {
 162                 return $self->{'last_sent_type'} = $type;
 163         }
 164         return $self->{'last_sent_type'};
 165 }
 166
 167 # When we're a client and we want to connect to a remote service
 168 # create( $app, username => $user, secret => $passwd );
 169 #    OR
 170 # create( $app, sysname => $user, secret => $shared_secret );
 171 sub create {
 172         my $class = shift;
 173         $class = ref($class) || $class;
 174
 175         my $app = shift;
 176         #my %auth_args = @_;
 177
 178
 179 #       unless ( $app &&
 180 #                exists($auth_args{secret}) &&
 181 #                ( exists($auth_args{username}) ||
 182 #                  exists($auth_args{sysname}) ) ) {
 183 #               throw OpenSRF::EX::User ( 'Insufficient authentication information for session creation');
 184 #       }
 185
 186         if( my $thingy = OpenSRF::AppSession->find_client( $app ) ) {
 187                         $logger->debug( "AppSession returning existing client session for $app", DEBUG );
 188                         return $thingy;
 189         } else {
 190                 $logger->debug( "AppSession creating new client session for $app", DEBUG );
 191         }
 192
 193
 194
 195         #my $auth = OpenSRF::DOM::Element::userAuth->new( %auth_args );
 196
 197         my $conf = OpenSRF::Utils::Config->current;
 198
 199         if( ! $conf ) {
 200                 OpenSRF::EX::Config->throw( "No suitable config found" );
 201         }
 202
 203         my $sess_id = time . rand( $$ );
 204         while ( $class->find($sess_id) ) {
 205                 $sess_id = time . rand( $$ );
 206         }
 207
 208         my $r_id = $conf->$app->transport_target ||
 209                         die("No remote id for $app!");
 210
 211         my $peer_handle = OpenSRF::Transport::PeerHandle->retrieve("client");
 212         if( ! $peer_handle ) {
 213                 $peer_handle = OpenSRF::Transport::PeerHandle->retrieve("system_client");
 214         }
 215
 216         my $self = bless { app_name    => $app,
 217                                 #client_auth => $auth,
 218                            #recv_queue  => [],
 219                            request_queue  => [],
 220                            endpoint    => CLIENT,
 221                            state       => DISCONNECTED,#since we're init'ing
 222                            session_id  => $sess_id,
 223                            remote_id   => $r_id,
 224                            api_level   => 1,
 225                            orig_remote_id   => $r_id,
 226                                 peer_handle => $peer_handle,
 227                                 session_threadTrace => 0,
 228                          } => $class;
 229
 230         $self->client_cache();
 231         $_CACHE{$sess_id} = $self;
 232         return $self->find_client( $app );
 233 }
 234
 235 sub api_level {
 236         return shift()->{api_level};
 237 }
 238
 239 sub app {
 240         return shift()->{app_name};
 241 }
 242
 243 sub reset {
 244         my $self = shift;
 245         $self->remote_id($$self{orig_remote_id});
 246 }
 247
 248 # 'connect' can be used as a constructor if called as a class method,
 249 # or used to connect a session that has disconnectd if called against
 250 # an existing session that seems to be disconnected, or was just built
 251 # using 'create' above.
 252
 253 # connect( $app, username => $user, secret => $passwd );
 254 #    OR
 255 # connect( $app, sysname => $user, secret => $shared_secret );
 256
 257 # --- Returns undef if the connect attempt times out.
 258 # --- Returns the OpenSRF::EX object if one is returned by the server
 259 # --- Returns self if connected
 260 sub connect {
 261         my $self = shift;
 262         my $class = ref($self) || $self;
 263
 264         if ( ref( $self ) and  $self->state && $self->state == CONNECTED  ) {
 265                 $logger->transport("ABC AppSession already connected", DEBUG );
 266         } else {
 267                 $logger->transport("ABC AppSession not connected, connecting..", DEBUG );
 268         }
 269         return $self if ( ref( $self ) and  $self->state && $self->state == CONNECTED  );
 270
 271         my $app = shift;
 272         my $api_level = shift;
 273         $api_level = 1 unless (defined $api_level);
 274
 275         $self = $class->create($app, @_) if (!ref($self));
 276         return undef unless ($self);
 277
 278         $self->{api_level} = $api_level;
 279
 280         $self->reset;
 281         $self->state(CONNECTING);
 282         $self->send('CONNECT', "");
 283
 284         my $time_remaining = OpenSRF::Utils::Config->current->client->connect_timeout;
 285
 286         while ( $self->state != CONNECTED  and $time_remaining > 0 ) {
 287                 my $starttime = time;
 288                 $self->queue_wait($time_remaining);
 289                 my $endtime = time;
 290                 $time_remaining -= ($endtime - $starttime);
 291         }
 292
 293         return undef unless($self->state == CONNECTED);
 294
 295         return $self;
 296 }
 297
 298 sub finish {
 299         my $self = shift;
 300         if( ! $self->session_id ) {
 301                 return 0;
 302         }
 303         #$self->disconnect if ($self->endpoint == CLIENT);
 304         for my $ses ( @_CLIENT_CACHE ) {
 305                 if ($ses->[2]->session_id eq $self->session_id) {
 306                         $ses->[1] = 1;
 307                 }
 308         }
 309 }
 310
 311 sub kill_me {
 312         my $self = shift;
 313         if( ! $self->session_id ) { return 0; }
 314         $self->disconnect;
 315         $logger->transport( "AppSession killing self: " . $self->session_id(), DEBUG );
 316         my @a;
 317         for my $ses ( @_CLIENT_CACHE ) {
 318                 push @a, $ses
 319                         if ($ses->[2]->session_id ne $self->session_id);
 320         }
 321         @_CLIENT_CACHE = @a;
 322         delete $_CACHE{$self->session_id};
 323         delete($$self{$_}) for (keys %$self);
 324 }
 325
 326 sub disconnect {
 327         my $self = shift;
 328         unless( $self->state == DISCONNECTED ) {
 329                 $self->send('DISCONNECT', "") if ($self->endpoint == CLIENT);;
 330                 $self->state( DISCONNECTED );
 331         }
 332         $self->reset;
 333 }
 334
 335 sub request {
 336         my $self = shift;
 337         my $meth = shift;
 338
 339         my $method;
 340         if (!ref $meth) {
 341                 $method = new OpenSRF::DomainObject::oilsMethod ( method => $meth );
 342         } else {
 343                 $method = $meth;
 344         }
 345
 346         $method->params( @_ );
 347
 348         $self->send('REQUEST',$method);
 349 }
 350
 351
 352 sub send {
 353         my $self = shift;
 354         my @payload_list = @_; # this is a Domain Object
 355
 356
 357         $logger->debug( "In send", INTERNAL );
 358
 359         my $tT;
 360
 361         if( @payload_list % 2 ) { $tT = pop @payload_list; }
 362
 363         if( ! @payload_list ) {
 364                 $logger->debug( "payload_list param is incomplete in AppSession::send()", ERROR );
 365                 return undef;
 366         }
 367
 368         my $doc = OpenSRF::DOM->createDocument();
 369
 370         $logger->debug( "In send2", INTERNAL );
 371
 372         my $disconnect = 0;
 373         my $connecting = 0;
 374
 375         while( @payload_list ) {
 376
 377                 my ($msg_type, $payload) = ( shift(@payload_list), shift(@payload_list) );
 378
 379                 if ($msg_type eq 'DISCONNECT' ) {
 380                         $disconnect++;
 381                         if( $self->state == DISCONNECTED) {
 382                                 next;
 383                         }
 384                 }
 385
 386                 if( $msg_type eq "CONNECT" ) { $connecting++; }
 387
 388
 389                 if( $payload ) {
 390                         $logger->debug( "Payload is ".$payload->toString, INTERNAL );
 391                 }
 392
 393
 394                 my $msg = OpenSRF::DomainObject::oilsMessage->new();
 395                 $logger->debug( "AppSession after creating oilsMessage $msg", INTERNAL );
 396
 397                 $msg->type($msg_type);
 398                 $logger->debug( "AppSession after adding type" . $msg->toString(), INTERNAL );
 399
 400                 #$msg->userAuth($self->client_auth) if ($self->endpoint == CLIENT && $msg_type eq 'CONNECT');
 401
 402                 no warnings;
 403                 $msg->threadTrace( $tT || int($self->session_threadTrace) || int($self->last_threadTrace) );
 404                 use warnings;
 405
 406                 if ($msg->type eq 'REQUEST') {
 407                         if ( !defined($tT) || $self->last_threadTrace != $tT ) {
 408                                 $msg->update_threadTrace;
 409                                 $self->session_threadTrace( $msg->threadTrace );
 410                                 $tT = $self->session_threadTrace;
 411                                 OpenSRF::AppRequest->new($self, $payload);
 412                         }
 413                 }
 414
 415                 $msg->api_level($self->api_level);
 416                 $msg->payload($payload) if $payload;
 417
 418                 $doc->documentElement->appendChild( $msg );
 419
 420
 421                 $logger->debug( "AppSession sending ".$msg->type." to ".$self->remote_id.
 422                         " with threadTrace [".$msg->threadTrace."]", INFO );
 423
 424         }
 425
 426         if ($self->endpoint == CLIENT and ! $disconnect) {
 427                 $self->queue_wait(0);
 428                 unless ($self->state == CONNECTED || ($self->state == CONNECTING && $connecting )) {
 429                         my $v = $self->connect();
 430                         if( ! $v ) {
 431                                 $logger->debug( "Unable to connect to remote service in AppSession::send()", ERROR );
 432                                 return undef;
 433                         }
 434                         if( $v and $v->can("class") and $v->class->isa( "OpenSRF::EX" ) ) {
 435                                 return $v;
 436                         }
 437                 }
 438         }
 439
 440         $logger->debug( "AppSession sending doc: " . $doc->toString(), INTERNAL );
 441
 442
 443         $self->{peer_handle}->send(
 444                                         to     => $self->remote_id,
 445                                    thread => $self->session_id,
 446                                    body   => $doc->toString );
 447
 448         return $self->app_request( $tT );
 449 }
 450
 451 sub app_request {
 452         my $self = shift;
 453         my $tT = shift;
 454
 455         return undef unless (defined $tT);
 456         my ($req) = grep { $_->threadTrace == $tT } @{ $self->{request_queue} };
 457
 458         return $req;
 459 }
 460
 461 sub remove_app_request {
 462         my $self = shift;
 463         my $req = shift;
 464
 465         my @list = grep { $_->threadTrace != $req->threadTrace } @{ $self->{request_queue} };
 466
 467         $self->{request_queue} = \@list;
 468 }
 469
 470 sub endpoint {
 471         return $_[0]->{endpoint};
 472 }
 473
 474
 475 sub session_id {
 476         my $self = shift;
 477         return $self->{session_id};
 478 }
 479
 480 sub push_queue {
 481         my $self = shift;
 482         my $resp = shift;
 483         my $req = $self->app_request($resp->[1]);
 484         return $req->push_queue( $resp->[0] ) if ($req);
 485         push @{ $self->{recv_queue} }, $resp->[0];
 486 }
 487
 488 sub last_threadTrace {
 489         my $self = shift;
 490         my $new_last_threadTrace = shift;
 491
 492         my $old_last_threadTrace = $self->{last_threadTrace};
 493         if (defined $new_last_threadTrace) {
 494                 $self->{last_threadTrace} = $new_last_threadTrace;
 495                 return $new_last_threadTrace unless ($old_last_threadTrace);
 496         }
 497
 498         return $old_last_threadTrace;
 499 }
 500
 501 sub session_threadTrace {
 502         my $self = shift;
 503         my $new_last_threadTrace = shift;
 504
 505         my $old_last_threadTrace = $self->{session_threadTrace};
 506         if (defined $new_last_threadTrace) {
 507                 $self->{session_threadTrace} = $new_last_threadTrace;
 508                 return $new_last_threadTrace unless ($old_last_threadTrace);
 509         }
 510
 511         return $old_last_threadTrace;
 512 }
 513
 514 sub last_message_type {
 515         my $self = shift;
 516         my $new_last_message_type = shift;
 517
 518         my $old_last_message_type = $self->{last_message_type};
 519         if (defined $new_last_message_type) {
 520                 $self->{last_message_type} = $new_last_message_type;
 521                 return $new_last_message_type unless ($old_last_message_type);
 522         }
 523
 524         return $old_last_message_type;
 525 }
 526
 527 sub last_message_api_level {
 528         my $self = shift;
 529         my $new_last_message_api_level = shift;
 530
 531         my $old_last_message_api_level = $self->{last_message_api_level};
 532         if (defined $new_last_message_api_level) {
 533                 $self->{last_message_api_level} = $new_last_message_api_level;
 534                 return $new_last_message_api_level unless ($old_last_message_api_level);
 535         }
 536
 537         return $old_last_message_api_level;
 538 }
 539
 540 sub remote_id {
 541         my $self = shift;
 542         my $new_remote_id = shift;
 543
 544         my $old_remote_id = $self->{remote_id};
 545         if (defined $new_remote_id) {
 546                 $self->{remote_id} = $new_remote_id;
 547                 return $new_remote_id unless ($old_remote_id);
 548         }
 549
 550         return $old_remote_id;
 551 }
 552
 553 sub client_auth {
 554         return undef;
 555         my $self = shift;
 556         my $new_ua = shift;
 557
 558         my $old_ua = $self->{client_auth};
 559         if (defined $new_ua) {
 560                 $self->{client_auth} = $new_ua;
 561                 return $new_ua unless ($old_ua);
 562         }
 563
 564         return $old_ua->cloneNode(1);
 565 }
 566
 567 sub state {
 568         my $self = shift;
 569         my $new_state = shift;
 570
 571         my $old_state = $self->{state};
 572         if (defined $new_state) {
 573                 $self->{state} = $new_state;
 574                 return $new_state unless ($old_state);
 575         }
 576
 577         return $old_state;
 578 }
 579
 580 sub DESTROY {
 581         my $self = shift;
 582         delete $$self{$_} for keys %$self;
 583         return undef;
 584 }
 585
 586 sub recv {
 587         my $self = shift;
 588         my @proto_args = @_;
 589         my %args;
 590
 591         if ( @proto_args ) {
 592                 if ( !(@proto_args % 2) ) {
 593                         %args = @proto_args;
 594                 } elsif (@proto_args == 1) {
 595                         %args = ( timeout => @proto_args );
 596                 }
 597         }
 598
 599         #$logger->debug( ref($self). " recv_queue before wait: " . $self->_print_queue(), INTERNAL );
 600
 601         if( exists( $args{timeout} ) ) {
 602                 $args{timeout} = int($args{timeout});
 603         } else {
 604                 $args{timeout} = 10;
 605         }
 606
 607         #$args{timeout} = 0 if ($self->complete);
 608
 609         $logger->debug( ref($self) ."->recv with timeout " . $args{timeout}, INTERNAL );
 610
 611         $args{count} ||= 1;
 612
 613         my $avail = @{ $self->{recv_queue} };
 614         my $time_remaining = $args{timeout};
 615
 616         while ( $avail < $args{count} and $time_remaining > 0 ) {
 617                 last if $self->complete;
 618                 my $starttime = time;
 619                 $self->queue_wait($time_remaining);
 620                 my $endtime = time;
 621                 $time_remaining -= ($endtime - $starttime);
 622                 $avail = @{ $self->{recv_queue} };
 623         }
 624
 625         #$logger->debug( ref($self)." queue after wait: " . $self->_print_queue(), INTERNAL );
 626
 627         my @list;
 628         while ( my $msg = shift @{ $self->{recv_queue} } ) {
 629                 push @list, $msg;
 630                 last if (scalar(@list) >= $args{count});
 631         }
 632
 633 #       $self->{recv_queue} = [@unlist, @{ $self->{recv_queue} }];
 634         $logger->debug( "Number of matched responses: " . @list, DEBUG );
 635
 636         $self->queue_wait(0); # check for statuses
 637
 638         return $list[0] unless (wantarray);
 639         return @list;
 640 }
 641
 642 sub push_resend {
 643         my $self = shift;
 644         push @OpenSRF::AppSession::_RESEND_QUEUE, @_;
 645 }
 646
 647 sub flush_resend {
 648         my $self = shift;
 649         $logger->debug( "Resending..." . @_RESEND_QUEUE, DEBUG );
 650         while ( my $req = shift @OpenSRF::AppSession::_RESEND_QUEUE ) {
 651                 $req->resend;
 652         }
 653 }
 654
 655
 656 sub queue_wait {
 657         my $self = shift;
 658         if( ! $self->{peer_handle} ) { return 0; }
 659         my $timeout = shift || 0;
 660         $logger->debug( "Calling queue_wait($timeout)" , DEBUG );
 661         $logger->debug( "Timestamp before process($timeout) : " . $logger->format_time(), INTERNAL );
 662         my $o = $self->{peer_handle}->process($timeout);
 663         $logger->debug( "Timestamp after  process($timeout) : " . $logger->format_time(), INTERNAL );
 664         $self->flush_resend;
 665         return $o;
 666 }
 667
 668 sub _print_queue {
 669         my( $self ) = @_;
 670         my $string = "";
 671         foreach my $msg ( @{$self->{recv_queue}} ) {
 672                 $string = $string . $msg->toString(1) . "\n";
 673         }
 674         return $string;
 675 }
 676
 677 sub status {
 678         my $self = shift;
 679         $self->send( 'STATUS', @_ );
 680 }
 681
 682 #-------------------------------------------------------------------------------
 683
 684 package OpenSRF::AppRequest;
 685 use base qw/OpenSRF::AppSession/;
 686 use OpenSRF::Utils::Logger qw/:level/;
 687 use OpenSRF::DomainObject::oilsResponse qw/:status/;
 688
 689 sub new {
 690         my $class = shift;
 691         $class = ref($class) || $class;
 692
 693         my $session = shift;
 694         my $threadTrace = $session->session_threadTrace || $session->last_threadTrace;
 695         my $payload = shift;
 696
 697         my $self = {    session     => $session,
 698                         threadTrace => $threadTrace,
 699                         payload     => $payload,
 700                         complete    => 0,
 701                         recv_queue  => [],
 702         };
 703
 704         bless $self => $class;
 705
 706         push @{ $self->session->{request_queue} }, $self;
 707
 708         return $self;
 709 }
 710
 711 sub queue_size {
 712         my $size = @{$_[0]->{recv_queue}};
 713         return $size;
 714 }
 715
 716 sub send {
 717         shift()->session->send(@_);
 718 }
 719
 720 sub finish {
 721         my $self = shift;
 722         $self->session->remove_app_request($self);
 723         delete($$self{$_}) for (keys %$self);
 724 }
 725
 726 sub session {
 727         return shift()->{session};
 728 }
 729
 730 sub complete {
 731         my $self = shift;
 732         my $complete = shift;
 733         return $self->{complete} if ($self->{complete});
 734         if (defined $complete) {
 735                 $self->{complete} = $complete;
 736         } else {
 737                 $self->session->queue_wait(0);
 738         }
 739         return $self->{complete};
 740 }
 741
 742 sub wait_complete {
 743         my $self = shift;
 744         my $timeout = shift || 1;
 745         my $time_remaining = $timeout;
 746
 747         while ( ! $self->complete  and $time_remaining > 0 ) {
 748                 my $starttime = time;
 749                 $self->queue_wait($time_remaining);
 750                 my $endtime = time;
 751                 $time_remaining -= ($endtime - $starttime);
 752         }
 753
 754         return $self->complete;
 755 }
 756
 757 sub threadTrace {
 758         return shift()->{threadTrace};
 759 }
 760
 761 sub push_queue {
 762         my $self = shift;
 763         my $resp = shift;
 764         if( !$resp ) { return 0; }
 765         push @{ $self->{recv_queue} }, $resp;
 766         OpenSRF::Utils::Logger->debug( "AppRequest pushed ".$resp->toString(), INTERNAL );
 767 }
 768
 769 sub queue_wait {
 770         my $self = shift;
 771         OpenSRF::Utils::Logger->debug( "Calling queue_wait(@_)", DEBUG );
 772         return $self->session->queue_wait(@_)
 773 }
 774
 775 sub payload { return shift()->{payload}; }
 776
 777 sub resend {
 778         my $self = shift;
 779         OpenSRF::Utils::Logger->debug(
 780                 "I'm resending the request for threadTrace ". $self->threadTrace, DEBUG);
 781         if($self->payload) {
 782                 OpenSRF::Utils::Logger->debug($self->payload->toString,INTERNAL);
 783         }
 784         return $self->session->send('REQUEST', $self->payload, $self->threadTrace );
 785 }
 786
 787 sub status {
 788         my $self = shift;
 789         my $msg = shift;
 790         $self->session->send( 'STATUS',$msg, $self->threadTrace );
 791 }
 792
 793 sub respond {
 794         my $self = shift;
 795         my $msg = shift;
 796
 797         my $response;
 798         if (ref($msg) && $msg->can('getAttribute') && $msg->getAttribute('name') =~ /oilsResult/) {
 799                 $response = $msg;
 800         } else {
 801                 $response = new OpenSRF::DomainObject::oilsResult;
 802                 $response->content($msg);
 803         }
 804
 805         $self->session->send('RESULT', $response, $self->threadTrace);
 806 }
 807
 808 sub respond_complete {
 809         my $self = shift;
 810         my $msg = shift;
 811
 812         my $response;
 813         if (ref($msg) && $msg->can('getAttribute') && $msg->getAttribute('name') =~ /oilsResult/) {
 814                 $response = $msg;
 815         } else {
 816                 $response = new OpenSRF::DomainObject::oilsResult;
 817                 $response->content($msg);
 818         }
 819
 820         my $stat = OpenSRF::DomainObject::oilsConnectStatus->new(
 821                 statusCode => STATUS_COMPLETE(),
 822                 status => 'Request Complete' );
 823
 824         $self->session->send( 'RESULT' => $response, 'STATUS' => $stat, $self->threadTrace);
 825
 826 }
 827
 828
 829 package OpenSRF::AppSubrequest;
 830
 831 sub respond {
 832         my $self = shift;
 833         my $resp = shift;
 834         push @{$$self{resp}}, $resp;
 835 }
 836
 837 sub new {
 838         return bless({resp => []}, 'OpenSRF::AppSubrequest');
 839 }
 840
 841 sub responses { @{$_[0]->{resp}} }
 842
 843 sub status {}
 844
 845
 846 1;
 847