src/perl/lib/OpenSRF/AppSession.pm

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