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