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         $0 = "$0*";
  71
  72
  73         if( ! $data or ! defined( $data ) or $data eq "" ) {
  74                 throw OpenSRF::EX::Socket(
  75                                 "Unix child received empty data from socket" );
  76         }
  77
  78         if( ! close( $self->{server}->{client} ) ) {
  79                 $logger->debug( "Error closing Unix socket: $!", ERROR );
  80         }
  81
  82
  83         my $app = $self->app();
  84         $logger->transport( "UnixServer for $app received $data", INTERNAL );
  85
  86         my $app_session = OpenSRF::Transport->handler( $self->app(), $data );
  87
  88         if(!ref($app_session)) {
  89                 $logger->transport( "Did not receive AppSession from transport handler, returning...", WARN );
  90                 $0 =~ s/\*//g;
  91                 return;
  92         }
  93
  94         my $client = OpenSRF::Utils::SettingsClient->new();
  95         my $keepalive = $client->config_value("apps", $self->app(), "keepalive");
  96
  97         my $req_counter = 0;
  98         while( $app_session->state and $app_session->state != $app_session->DISCONNECTED() and
  99                         $app_session->find( $app_session->session_id ) ) {
 100
 101
 102                 my $before = time;
 103                 $logger->transport( "UnixServer calling queue_wait $keepalive", INTERNAL );
 104                 $app_session->queue_wait( $keepalive );
 105                 $logger->transport( "after queue wait $keepalive", INTERNAL );
 106                 my $after = time;
 107
 108                 if( ($after - $before) >= $keepalive ) {
 109
 110                         my $res = OpenSRF::DomainObject::oilsConnectStatus->new(
 111                                                                         status => "Disconnected on timeout",
 112                                                                         statusCode => STATUS_TIMEOUT);
 113                         $app_session->status($res);
 114                         $app_session->state( $app_session->DISCONNECTED() );
 115                         last;
 116                 }
 117
 118         }
 119
 120         my $x = 0;
 121         while( 1 ) {
 122                 $logger->transport( "Looping on zombies " . $x++ , DEBUG);
 123                 last unless ( $app_session->queue_wait(0));
 124         }
 125
 126         $logger->transport( "Timed out, disconnected, or auth failed", INFO );
 127         $app_session->kill_me;
 128
 129         $0 =~ s/\*//g;
 130
 131
 132 }
 133
 134
 135 sub serve {
 136         my( $self ) = @_;
 137
 138         my $app = $self->app();
 139
 140         my $client = OpenSRF::Utils::SettingsClient->new();
 141         $logger->transport("Max Req: " . $client->config_value("apps", $app, "unix_config", "max_requests" ), INFO );
 142
 143         my $min_servers = $client->config_value("apps", $app, "unix_config", "min_children" );
 144         my $max_servers = $client->config_value("apps", $app, "unix_config", "max_children" );
 145         my $min_spare    =      $client->config_value("apps", $app, "unix_config", "min_spare_children" );
 146         my $max_spare    = $client->config_value("apps", $app, "unix_config", "max_spare_children" );
 147         my $max_requests = $client->config_value("apps", $app, "unix_config", "max_requests" );
 148         my $log_file = join("/", $client->config_value("dirs", "log"),
 149                                 $client->config_value("apps", $app, "unix_config", "unix_log" ));
 150         my $port =      join("/", $client->config_value("dirs", "sock"),
 151                                 $client->config_value("apps", $app, "unix_config", "unix_sock" ));
 152         my $pid_file =  join("/", $client->config_value("dirs", "pid"),
 153                                 $client->config_value("apps", $app, "unix_config", "unix_pid" ));
 154
 155
 156
 157 =head
 158         $self->{server}->{min_severs} = $min_servers;
 159         $self->{server}->{max_severs} = $max_servers;
 160         $self->{server}->{min_spare_severs} = $min_spare;
 161         $self->{server}->{max_spare_servers} = $max_spare;
 162         $self->{server}->{max_request} = $max_requests;
 163         $self->{server}->{log_file} = $log_file;
 164         $self->{server}->{pid_file} = $pid_file;
 165         $self->{server}->{log_level} = 4;
 166         $self->{server}->{proto} = "unix";
 167 =cut
 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
 181 }
 182
 183 sub configure_hook {
 184         my $self = shift;
 185         my $app = $self->app;
 186
 187         $logger->debug( "Setting application implementaion for $app", DEBUG );
 188         my $client = OpenSRF::Utils::SettingsClient->new();
 189         my $imp = $client->config_value("apps", $app, "implementation");
 190         OpenSRF::Application::server_class($app);
 191         OpenSRF::Application->application_implementation( $imp );
 192         JSON->register_class_hint( name => $imp, hint => $app, type => "hash" );
 193         OpenSRF::Application->application_implementation->initialize()
 194                 if (OpenSRF::Application->application_implementation->can('initialize'));
 195
 196         if( $client->config_value("server_type") !~ /fork/i  ) {
 197                 $self->child_init_hook();
 198         }
 199
 200         return OpenSRF::Application->application_implementation;
 201 }
 202
 203 sub child_finish_hook {
 204         my $self = shift;
 205         OpenSRF::AppSession->kill_client_session_cache;
 206 }
 207
 208 sub child_init_hook {
 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         my $peer_handle = OpenSRF::System::bootstrap_client("system_client");
 218         OpenSRF::Application->application_implementation->child_init
 219                 if (OpenSRF::Application->application_implementation->can('child_init'));
 220         return $peer_handle;
 221
 222 }
 223
 224 1;
 225