OpenSRF/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);
   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 JSON;
  14 use vars qw/@ISA $app/;
  15 use Carp;
  16
  17 # XXX Need to add actual logging statements in the code
  18 my $logger = "OpenSRF::Utils::Logger";
  19
  20 sub DESTROY { confess "Dying $$"; }
  21
  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
  72         my $orig = $0;
  73         $0 = "$0*";
  74
  75         if( ! $data or ! defined( $data ) or $data eq "" ) {
  76                 close($self->{server}->{client});
  77                 $logger->debug("Unix child received empty data from socket", ERROR);
  78                 $0 = $orig;
  79                 return;
  80         }
  81
  82
  83         if( ! close( $self->{server}->{client} ) ) {
  84                 $logger->debug( "Error closing Unix socket: $!", ERROR );
  85         }
  86
  87
  88         my $app = $self->app();
  89         $logger->transport( "UnixServer for $app received $data", INTERNAL );
  90
  91         my $app_session = OpenSRF::Transport->handler( $self->app(), $data );
  92
  93         if(!ref($app_session)) {
  94                 $logger->transport( "Did not receive AppSession from transport handler, returning...", WARN );
  95                 $0 = $orig;
  96                 return;
  97         }
  98
  99         if($app_session->stateless and $app_session->state != $app_session->CONNECTED()){
 100                 $logger->debug("Exiting keepalive for stateless session / orig = $orig");
 101                 $app_session->kill_me;
 102                 $0 = $orig;
 103                 return;
 104         }
 105
 106
 107         my $client = OpenSRF::Utils::SettingsClient->new();
 108         my $keepalive = $client->config_value("apps", $self->app(), "keepalive");
 109
 110         my $req_counter = 0;
 111         while( $app_session and
 112                         $app_session->state and
 113                         $app_session->state != $app_session->DISCONNECTED() and
 114                         $app_session->find( $app_session->session_id ) ) {
 115
 116
 117                 my $before = time;
 118                 $logger->debug( "UnixServer calling queue_wait $keepalive", INTERNAL );
 119                 $app_session->queue_wait( $keepalive );
 120                 $logger->debug( "after queue wait $keepalive", INTERNAL );
 121                 my $after = time;
 122
 123                 if( ($after - $before) >= $keepalive ) {
 124
 125                         my $res = OpenSRF::DomainObject::oilsConnectStatus->new(
 126                                                                         status => "Disconnected on timeout",
 127                                                                         statusCode => STATUS_TIMEOUT);
 128                         $app_session->status($res);
 129                         $app_session->state( $app_session->DISCONNECTED() );
 130                         last;
 131                 }
 132
 133         }
 134
 135         my $x = 0;
 136         while( $app_session && $app_session->queue_wait(0) ) {
 137                 $logger->debug( "Looping on zombies " . $x++ , DEBUG);
 138         }
 139
 140         $logger->debug( "Timed out, disconnected, or auth failed", INFO );
 141         $app_session->kill_me if ($app_session);
 142
 143         $0 = $orig;
 144
 145
 146 }
 147
 148
 149 sub serve {
 150         my( $self ) = @_;
 151
 152         my $app = $self->app();
 153
 154         $0 = "OpenSRF master [$app]";
 155
 156         system("rm -f /tmp/opensrf_unix_$app*");
 157
 158         my $client = OpenSRF::Utils::SettingsClient->new();
 159         $logger->transport("Max Req: " . $client->config_value("apps", $app, "unix_config", "max_requests" ), INFO );
 160
 161         my $min_servers = $client->config_value("apps", $app, "unix_config", "min_children" );
 162         my $max_servers = $client->config_value("apps", $app, "unix_config", "max_children" );
 163         my $min_spare    =      $client->config_value("apps", $app, "unix_config", "min_spare_children" );
 164         my $max_spare    = $client->config_value("apps", $app, "unix_config", "max_spare_children" );
 165         my $max_requests = $client->config_value("apps", $app, "unix_config", "max_requests" );
 166         my $log_file = join("/", $client->config_value("dirs", "log"),
 167                                 $client->config_value("apps", $app, "unix_config", "unix_log" ));
 168         my $port =      join("/", $client->config_value("dirs", "sock"),
 169                                 $client->config_value("apps", $app, "unix_config", "unix_sock" ));
 170         my $pid_file =  join("/", $client->config_value("dirs", "pid"),
 171                                 $client->config_value("apps", $app, "unix_config", "unix_pid" ));
 172
 173         my $file = "/tmp/" . "opensrf_unix_$app"."_" . time . rand( $$ ) . "_$$";
 174         my $file_string = "min_servers $min_servers\nmax_servers $max_servers\n" .
 175                 "min_spare_servers $min_spare\nmax_spare_servers $max_spare\n" .
 176                 "max_requests $max_requests\nlog_file $log_file\nproto unix\n" .
 177                 "port $port\npid_file $pid_file\nlog_level 3\n";
 178
 179         open F, "> $file" or die "Can't open $file : $!";
 180         print F $file_string;
 181         close F;
 182
 183         $self->run( 'conf_file' => $file );
 184         unlink($file);
 185
 186 }
 187
 188 sub configure_hook {
 189         my $self = shift;
 190         my $app = $self->app;
 191
 192         # boot a client
 193         OpenSRF::System->bootstrap_client( client_name => "system_client" );
 194
 195         $logger->debug( "Setting application implementaion for $app", DEBUG );
 196         my $client = OpenSRF::Utils::SettingsClient->new();
 197         my $imp = $client->config_value("apps", $app, "implementation");
 198         OpenSRF::Application::server_class($app);
 199         OpenSRF::Application->application_implementation( $imp );
 200         JSON->register_class_hint( name => $imp, hint => $app, type => "hash" );
 201         OpenSRF::Application->application_implementation->initialize()
 202                 if (OpenSRF::Application->application_implementation->can('initialize'));
 203
 204         if( $client->config_value("server_type") !~ /fork/i  ) {
 205                 $self->child_init_hook();
 206         }
 207
 208         my $con = OpenSRF::Transport::PeerHandle->retrieve;
 209         if($con) {
 210                 $con->disconnect;
 211         }
 212
 213         return OpenSRF::Application->application_implementation;
 214 }
 215
 216 sub child_finish_hook {
 217         my $self = shift;
 218         OpenSRF::AppSession->kill_client_session_cache;
 219 }
 220
 221 sub child_init_hook {
 222
 223         $0 =~ s/master/drone/g;
 224
 225         if ($ENV{OPENSRF_PROFILE}) {
 226                 my $file = $0;
 227                 $file =~ s/\W/_/go;
 228                 eval "use Devel::Profiler output_file => '/tmp/profiler_$file.out', buffer_size => 0;";
 229                 if ($@) {
 230                         $logger->debug("Could not load Devel::Profiler: $@",ERROR);
 231                 } else {
 232                         $0 .= ' [PROFILING]';
 233                         $logger->debug("Running under Devel::Profiler", INFO);
 234                 }
 235         }
 236
 237         my $self = shift;
 238
 239 #       $logger->transport(
 240 #                       "Creating PeerHandle from UnixServer child_init_hook", INTERNAL );
 241         OpenSRF::Transport::PeerHandle->construct( $self->app() );
 242         $logger->transport( "PeerHandle Created from UnixServer child_init_hook", INTERNAL );
 243
 244         OpenSRF::Application->application_implementation->child_init
 245                 if (OpenSRF::Application->application_implementation->can('child_init'));
 246         return OpenSRF::Transport::PeerHandle->retrieve;
 247
 248 }
 249
 250 1;
 251