src/perl/lib/OpenSRF/Transport.pm

   1 package OpenSRF::Transport;
   2 use strict; use warnings;
   3 use base 'OpenSRF';
   4 use Time::HiRes qw/time/;
   5 use OpenSRF::AppSession;
   6 use OpenSRF::Utils::JSON;
   7 use OpenSRF::Utils::Logger qw(:level);
   8 use OpenSRF::DomainObject::oilsResponse qw/:status/;
   9 use OpenSRF::EX qw/:try/;
  10 use OpenSRF::Transport::SlimJabber::MessageWrapper;
  11
  12 #------------------
  13 # --- These must be implemented by all Transport subclasses
  14 # -------------------------------------------
  15
  16 =head2 get_peer_client
  17
  18 Returns the name of the package responsible for client communication
  19
  20 =cut
  21
  22 sub get_peer_client { shift()->alert_abstract(); }
  23
  24 =head2 get_msg_envelope
  25
  26 Returns the name of the package responsible for parsing incoming messages
  27
  28 =cut
  29
  30 sub get_msg_envelope { shift()->alert_abstract(); }
  31
  32 # -------------------------------------------
  33
  34 our $message_envelope;
  35 my $logger = "OpenSRF::Utils::Logger";
  36
  37
  38
  39 =head2 message_envelope( [$envelope] );
  40
  41 Sets the message envelope class that will allow us to extract
  42 information from the messages we receive from the low
  43 level transport
  44
  45 =cut
  46
  47 sub message_envelope {
  48         my( $class, $envelope ) = @_;
  49         if( $envelope ) {
  50                 $message_envelope = $envelope;
  51                 $envelope->use;
  52                 if( $@ ) {
  53                         $logger->error(
  54                                         "Error loading message_envelope: $envelope -> $@", ERROR);
  55                 }
  56         }
  57         return $message_envelope;
  58 }
  59
  60 =head2 handler( $data )
  61
  62 Creates a new MessageWrapper, extracts the remote_id, session_id, and message body
  63 from the message.  Then, creates or retrieves the AppSession object with the session_id and remote_id.
  64 Finally, creates the message document from the body of the message and calls
  65 the handler method on the message document.
  66
  67 =cut
  68
  69 sub handler {
  70         my $start_time = time();
  71         my( $class, $service, $data ) = @_;
  72
  73         $logger->transport( "Transport handler() received $data", INTERNAL );
  74
  75         my $remote_id   = $data->from;
  76         my $sess_id     = $data->thread;
  77         my $body        = $data->body;
  78         my $type        = $data->type;
  79
  80         $logger->set_osrf_xid($data->osrf_xid);
  81
  82
  83         if (defined($type) and $type eq 'error') {
  84                 throw OpenSRF::EX::Session ("$remote_id IS NOT CONNECTED TO THE NETWORK!!!");
  85
  86         }
  87
  88         # See if the app_session already exists.  If so, make
  89         # sure the sender hasn't changed if we're a server
  90         my $app_session = OpenSRF::AppSession->find( $sess_id );
  91         if( $app_session and $app_session->endpoint == $app_session->SERVER() and
  92                         $app_session->remote_id ne $remote_id ) {
  93
  94             my $c = OpenSRF::Utils::SettingsClient->new();
  95         if($c->config_value("apps", $app_session->service, "migratable")) {
  96             $logger->debug("service is migratable, new client is $remote_id");
  97         } else {
  98
  99                     $logger->warn("Backend Gone or invalid sender");
 100                     my $res = OpenSRF::DomainObject::oilsBrokenSession->new();
 101                     $res->status( "Backend Gone or invalid sender, Reconnect" );
 102                     $app_session->status( $res );
 103                     return 1;
 104         }
 105         }
 106
 107         # Retrieve or build the app_session as appropriate (server_build decides which to do)
 108         $logger->transport( "AppSession is valid or does not exist yet", INTERNAL );
 109         $app_session = OpenSRF::AppSession->server_build( $sess_id, $remote_id, $service );
 110
 111         if( ! $app_session ) {
 112                 throw OpenSRF::EX::Session ("Transport::handler(): No AppSession object returned from server_build()");
 113         }
 114
 115         # Create a document from the JSON contained within the message
 116         my $doc;
 117         eval { $doc = OpenSRF::Utils::JSON->JSON2perl($body); };
 118         if( $@ ) {
 119
 120                 $logger->warn("Received bogus JSON: $@");
 121                 $logger->warn("Bogus JSON data: $body");
 122                 my $res = OpenSRF::DomainObject::oilsXMLParseError->new( status => "JSON Parse Error --- $body\n\n$@" );
 123
 124                 $app_session->status($res);
 125                 #$app_session->kill_me;
 126                 return 1;
 127         }
 128
 129         $logger->transport( "Transport::handler() creating \n$body", INTERNAL );
 130
 131         # We need to disconnect the session if we got a jabber error on the client side.  For
 132         # server side, we'll just tear down the session and go away.
 133         if (defined($type) and $type eq 'error') {
 134                 # If we're a server
 135                 if( $app_session->endpoint == $app_session->SERVER() ) {
 136                         $app_session->kill_me;
 137                         return 1;
 138                 } else {
 139                         $app_session->reset;
 140                         $app_session->state( $app_session->DISCONNECTED );
 141                         # below will lead to infinite looping, should return an exception
 142                         #$app_session->push_resend( $app_session->app_request(
 143                         #               $doc->documentElement->firstChild->threadTrace ) );
 144                         $logger->debug(
 145                                 "Got Jabber error on client connection $remote_id, nothing we can do..", ERROR );
 146                         return 1;
 147                 }
 148         }
 149
 150         # cycle through and pass each oilsMessage contained in the message
 151         # up to the message layer for processing.
 152         for my $msg (@$doc) {
 153
 154                 next unless (   $msg && UNIVERSAL::isa($msg => 'OpenSRF::DomainObject::oilsMessage'));
 155
 156                 OpenSRF::AppSession->ingress($msg->sender_ingress);
 157
 158                 if( $app_session->endpoint == $app_session->SERVER() ) {
 159
 160                         try {
 161
 162                                 if( ! $msg->handler( $app_session ) ) { return 0; }
 163                                 $logger->info(sprintf("Message processing duration: %.3f", (time() - $start_time)));
 164
 165                         } catch Error with {
 166
 167                                 my $e = shift;
 168                                 my $res = OpenSRF::DomainObject::oilsServerError->new();
 169                                 $res->status( $res->status . "\n$e");
 170                                 $app_session->status($res) if $res;
 171                                 $app_session->kill_me;
 172                                 return 0;
 173
 174                         };
 175
 176                 } else {
 177
 178                         if( ! $msg->handler( $app_session ) ) { return 0; }
 179                         $logger->debug(sub{return sprintf("Response processing duration: %.3f", (time() - $start_time)) });
 180
 181                 }
 182         }
 183
 184         return $app_session;
 185 }
 186
 187 1;