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