src/perlmods/OpenSRF/UnixServer.pm

   1 package OpenSRF::UnixServer;
   2 use strict; use warnings;
   3 use base qw/OpenSRF/;
   4 use OpenSRF::EX qw(:try);
   5 use OpenSRF::Utils::Logger qw(:level $logger);
   6 use OpenSRF::Transport::PeerHandle;
   7 use OpenSRF::Application;
   8 use OpenSRF::AppSession;
   9 use OpenSRF::DomainObject::oilsResponse qw/:status/;
  10 use OpenSRF::System;
  11 use OpenSRF::Utils::SettingsClient;
  12 use Time::HiRes qw(time);
  13 use OpenSRF::Utils::JSON;
  14 use vars qw/@ISA $app/;
  15 use Fcntl qw(F_GETFL F_SETFL O_NONBLOCK);
  16 use Carp;
  17
  18 use IO::Socket::INET;
  19 use IO::Socket::UNIX;
  20
  21 sub DESTROY { confess "Dying $$"; }
  22
  23 =head1 What am I
  24
  25 All inbound messages are passed on to the UnixServer for processing.
  26 We take the data, close the Unix socket, and pass the data on to our abstract
  27 'process()' method.
  28
  29 Our purpose is to 'multiplex' a single TCP connection into multiple 'client' connections.
  30 So when you pass data down the Unix socket to us, we have been preforked and waiting
  31 to disperse new data among us.
  32
  33 =cut
  34
  35 sub app { return $app; }
  36
  37 {
  38
  39         sub new {
  40                 my( $class, $app1 ) = @_;
  41                 if( ! $app1 ) {
  42                         throw OpenSRF::EX::InvalidArg( "UnixServer requires an app name to run" );
  43                 }
  44                 $app = $app1;
  45                 my $self = bless( {}, $class );
  46 #               my $client = OpenSRF::Utils::SettingsClient->new();
  47 #               if( $client->config_value("server_type") !~ /fork/i ||
  48 #                               OpenSRF::Utils::Config->current->bootstrap->settings_config ) {
  49 #                       warn "Calling hooks for non-prefork\n";
  50 #                       $self->configure_hook();
  51 #                       $self->child_init_hook();
  52 #               }
  53                 return $self;
  54         }
  55
  56 }
  57
  58 =head2 process_request()
  59
  60 Takes the incoming data, closes the Unix socket and hands the data untouched
  61 to the abstract process() method.  This method is implemented in our subclasses.
  62
  63 =cut
  64
  65 sub process_request {
  66
  67         my $self = shift;
  68         my $data; my $d;
  69         while( $d = <STDIN> ) { $data .= $d; }
  70
  71         my $orig = $0;
  72         $0 = "$0*";
  73
  74         if( ! $data or ! defined( $data ) or $data eq "" ) {
  75                 close($self->{server}->{client});
  76                 $logger->debug("Unix child received empty data from socket", ERROR);
  77                 $0 = $orig;
  78                 return;
  79         }
  80
  81
  82         if( ! close( $self->{server}->{client} ) ) {
  83                 $logger->debug( "Error closing Unix socket: $!", ERROR );
  84         }
  85
  86         my $app = $self->app();
  87         $logger->transport( "UnixServer for $app received $data", INTERNAL );
  88
  89         # --------------------------------------------------------------
  90         # Drop all data from the socket before coninuting to process
  91         # --------------------------------------------------------------
  92         my $ph = OpenSRF::Transport::PeerHandle->retrieve;
  93         if(!$ph->flush_socket()) {
  94                 $logger->error("We received a request ".
  95                         "and we are no longer connected to the jabber network. ".
  96                         "We will go away and drop this request: $data");
  97                 exit;
  98         }
  99
 100         my $app_session = OpenSRF::Transport->handler( $self->app(), $data );
 101
 102         if(!ref($app_session)) {
 103                 $logger->transport( "Did not receive AppSession from transport handler, returning...", WARN );
 104                 $0 = $orig;
 105                 return;
 106         }
 107
 108         if($app_session->stateless and $app_session->state != $app_session->CONNECTED()){
 109                 $logger->debug("Exiting keepalive for stateless session / orig = $orig");
 110                 $app_session->kill_me;
 111                 $0 = $orig;
 112                 return;
 113         }
 114
 115
 116         my $client = OpenSRF::Utils::SettingsClient->new();
 117         my $keepalive = $client->config_value("apps", $self->app(), "keepalive");
 118
 119         my $req_counter = 0;
 120         while( $app_session and
 121                         $app_session->state and
 122                         $app_session->state != $app_session->DISCONNECTED() and
 123                         $app_session->find( $app_session->session_id ) ) {
 124
 125
 126                 my $before = time;
 127                 $logger->debug( "UnixServer calling queue_wait $keepalive", INTERNAL );
 128                 $app_session->queue_wait( $keepalive );
 129                 $logger->debug( "after queue wait $keepalive", INTERNAL );
 130                 my $after = time;
 131
 132                 if( ($after - $before) >= $keepalive ) {
 133
 134                         my $res = OpenSRF::DomainObject::oilsConnectStatus->new(
 135                                                                         status => "Disconnected on timeout",
 136                                                                         statusCode => STATUS_TIMEOUT);
 137                         $app_session->status($res);
 138                         $app_session->state( $app_session->DISCONNECTED() );
 139                         last;
 140                 }
 141
 142         }
 143
 144         my $x = 0;
 145         while( $app_session && $app_session->queue_wait(0) ) {
 146                 $logger->debug( "Looping on zombies " . $x++ , DEBUG);
 147         }
 148
 149         $logger->debug( "Timed out, disconnected, or auth failed" );
 150         $app_session->kill_me if ($app_session);
 151
 152         $0 = $orig;
 153 }
 154
 155
 156 sub serve {
 157         my( $self ) = @_;
 158
 159         my $app = $self->app();
 160         $logger->set_service($app);
 161
 162         $0 = "OpenSRF master [$app]";
 163
 164         my $client = OpenSRF::Utils::SettingsClient->new();
 165     my @base = ('apps', $app, 'unix_config' );
 166
 167         my $min_servers = $client->config_value(@base, 'min_children');
 168         my $max_servers = $client->config_value(@base, "max_children" );
 169         my $min_spare = $client->config_value(@base, "min_spare_children" );
 170         my $max_spare = $client->config_value(@base, "max_spare_children" );
 171         my $max_requests = $client->config_value(@base, "max_requests" );
 172     # fwiw, these file paths are (obviously) not portable
 173         my $log_file = join("/", $client->config_value("dirs", "log"), $client->config_value(@base, "unix_log" ));
 174         my $port = join("/", $client->config_value("dirs", "sock"), $client->config_value(@base, "unix_sock" ));
 175         my $pid_file = join("/", $client->config_value("dirs", "pid"), $client->config_value(@base, "unix_pid" ));
 176
 177     $min_spare ||= $min_servers;
 178     $max_spare ||= $max_servers;
 179     $max_requests ||= 1000;
 180
 181     $logger->info("UnixServer: min=$min_servers, max=$max_servers, min_spare=$min_spare ".
 182         "max_spare=$max_spare, max_req=$max_requests, log_file=$log_file, port=$port, pid_file=$pid_file");
 183
 184     $self->run(
 185         min_servers => $min_servers,
 186         max_servers => $max_servers,
 187         min_spare_servers => $min_spare,
 188         max_spare_servers => $max_spare,
 189         max_requests => $max_requests,
 190         log_file => $log_file,
 191         port => $port,
 192         proto => 'unix',
 193         pid_file => $pid_file,
 194     );
 195
 196 }
 197
 198
 199 sub configure_hook {
 200         my $self = shift;
 201         my $app = $self->app;
 202
 203         # boot a client
 204         OpenSRF::System->bootstrap_client( client_name => "system_client" );
 205
 206         $logger->debug( "Setting application implementaion for $app", DEBUG );
 207         my $client = OpenSRF::Utils::SettingsClient->new();
 208         my $imp = $client->config_value("apps", $app, "implementation");
 209         OpenSRF::Application::server_class($app);
 210         OpenSRF::Application->application_implementation( $imp );
 211         OpenSRF::Utils::JSON->register_class_hint( name => $imp, hint => $app, type => "hash" );
 212         OpenSRF::Application->application_implementation->initialize()
 213                 if (OpenSRF::Application->application_implementation->can('initialize'));
 214
 215         if( $client->config_value("server_type") !~ /fork/i  ) {
 216                 $self->child_init_hook();
 217         }
 218
 219         my $con = OpenSRF::Transport::PeerHandle->retrieve;
 220         if($con) {
 221                 $con->disconnect;
 222         }
 223
 224         return OpenSRF::Application->application_implementation;
 225 }
 226
 227 sub child_init_hook {
 228
 229         $0 =~ s/master/drone/g;
 230
 231         if ($ENV{OPENSRF_PROFILE}) {
 232                 my $file = $0;
 233                 $file =~ s/\W/_/go;
 234                 eval "use Devel::Profiler output_file => '/tmp/profiler_$file.out', buffer_size => 0;";
 235                 if ($@) {
 236                         $logger->debug("Could not load Devel::Profiler: $@",ERROR);
 237                 } else {
 238                         $0 .= ' [PROFILING]';
 239                         $logger->debug("Running under Devel::Profiler", INFO);
 240                 }
 241         }
 242
 243         my $self = shift;
 244
 245 #       $logger->transport(
 246 #                       "Creating PeerHandle from UnixServer child_init_hook", INTERNAL );
 247         OpenSRF::Transport::PeerHandle->construct( $self->app() );
 248         $logger->transport( "PeerHandle Created from UnixServer child_init_hook", INTERNAL );
 249
 250         OpenSRF::Application->application_implementation->child_init
 251                 if (OpenSRF::Application->application_implementation->can('child_init'));
 252
 253         return OpenSRF::Transport::PeerHandle->retrieve;
 254 }
 255
 256 sub child_finish_hook {
 257     $logger->debug("attempting to call child exit handler...");
 258         OpenSRF::Application->application_implementation->child_exit
 259                 if (OpenSRF::Application->application_implementation->can('child_exit'));
 260 }
 261
 262
 263 1;
 264