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