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