src/perlmods/OpenSRF/UnixServer.pm

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