refactored the unix::server config handling by getting rid of the old temp config...
[OpenSRF.git] / 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 $logger);
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 Time::HiRes qw(time);
13 use JSON;
14 use vars qw/@ISA $app/;
15 use Fcntl qw(F_GETFL F_SETFL O_NONBLOCK);
16 use Carp;
17
18 use IO::Socket::INET;
19 use IO::Socket::UNIX;
20
21 sub DESTROY { confess "Dying $$"; }
22
23 =head1 What am I
24
25 All inbound messages are passed on to the UnixServer for processing.
26 We take the data, close the Unix socket, and pass the data on to our abstract
27 'process()' method.  
28
29 Our purpose is to 'multiplex' a single TCP connection into multiple 'client' connections.
30 So when you pass data down the Unix socket to us, we have been preforked and waiting
31 to disperse new data among us.
32
33 =cut
34
35 sub app { return $app; }
36
37 {
38
39         sub new {
40                 my( $class, $app1 ) = @_;
41                 if( ! $app1 ) {
42                         throw OpenSRF::EX::InvalidArg( "UnixServer requires an app name to run" );
43                 }
44                 $app = $app1;
45                 my $self = bless( {}, $class );
46 #               my $client = OpenSRF::Utils::SettingsClient->new();
47 #               if( $client->config_value("server_type") !~ /fork/i || 
48 #                               OpenSRF::Utils::Config->current->bootstrap->settings_config ) {
49 #                       warn "Calling hooks for non-prefork\n";
50 #                       $self->configure_hook();
51 #                       $self->child_init_hook();
52 #               }
53                 return $self;
54         }
55
56 }
57
58 =head2 process_request()
59
60 Takes the incoming data, closes the Unix socket and hands the data untouched 
61 to the abstract process() method.  This method is implemented in our subclasses.
62
63 =cut
64
65 sub process_request {
66
67         my $self = shift;
68         my $data; my $d;
69         while( $d = <STDIN> ) { $data .= $d; }
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         my $app = $self->app();
87         $logger->transport( "UnixServer for $app received $data", INTERNAL );
88
89         # --------------------------------------------------------------
90         # Drop all data from the socket before coninuting to process
91         # --------------------------------------------------------------
92         my $ph = OpenSRF::Transport::PeerHandle->retrieve;
93         if(!$ph->flush_socket()) {
94                 $logger->error("We received a request ".
95                         "and we are no longer connected to the jabber network. ".
96                         "We will go away and drop this request: $data");
97                 exit;
98         }
99
100         my $app_session = OpenSRF::Transport->handler( $self->app(), $data );
101
102         if(!ref($app_session)) {
103                 $logger->transport( "Did not receive AppSession from transport handler, returning...", WARN );
104                 $0 = $orig;
105                 return;
106         }
107
108         if($app_session->stateless and $app_session->state != $app_session->CONNECTED()){
109                 $logger->debug("Exiting keepalive for stateless session / orig = $orig");
110                 $app_session->kill_me;
111                 $0 = $orig;
112                 return;
113         }
114
115
116         my $client = OpenSRF::Utils::SettingsClient->new();
117         my $keepalive = $client->config_value("apps", $self->app(), "keepalive");
118
119         my $req_counter = 0;
120         while( $app_session and 
121                         $app_session->state and 
122                         $app_session->state != $app_session->DISCONNECTED() and
123                         $app_session->find( $app_session->session_id ) ) {
124                 
125
126                 my $before = time;
127                 $logger->debug( "UnixServer calling queue_wait $keepalive", INTERNAL );
128                 $app_session->queue_wait( $keepalive );
129                 $logger->debug( "after queue wait $keepalive", INTERNAL );
130                 my $after = time;
131
132                 if( ($after - $before) >= $keepalive ) { 
133
134                         my $res = OpenSRF::DomainObject::oilsConnectStatus->new(
135                                                                         status => "Disconnected on timeout",
136                                                                         statusCode => STATUS_TIMEOUT);
137                         $app_session->status($res);
138                         $app_session->state( $app_session->DISCONNECTED() );
139                         last;
140                 }
141         
142         }
143
144         my $x = 0;
145         while( $app_session && $app_session->queue_wait(0) ) {
146                 $logger->debug( "Looping on zombies " . $x++ , DEBUG);
147         }
148
149         $logger->debug( "Timed out, disconnected, or auth failed" );
150         $app_session->kill_me if ($app_session);
151
152         $0 = $orig;
153 }
154
155
156 sub serve {
157         my( $self ) = @_;
158
159         my $app = $self->app();
160         $logger->set_service($app);
161
162         $0 = "OpenSRF master [$app]";
163
164         my $client = OpenSRF::Utils::SettingsClient->new();
165     my @base = ('apps', $app, 'unix_config' );
166
167         my $min_servers = $client->config_value(@base, 'min_children');
168         my $max_servers = $client->config_value(@base, "max_children" );
169         my $min_spare = $client->config_value(@base, "min_spare_children" );
170         my $max_spare = $client->config_value(@base, "max_spare_children" );
171         my $max_requests = $client->config_value(@base, "max_requests" );
172     # fwiw, these file paths are (obviously) not portable
173         my $log_file = join("/", $client->config_value("dirs", "log"), $client->config_value(@base, "unix_log" ));
174         my $port = join("/", $client->config_value("dirs", "sock"), $client->config_value(@base, "unix_sock" ));
175         my $pid_file = join("/", $client->config_value("dirs", "pid"), $client->config_value(@base, "unix_pid" ));
176
177     $min_spare ||= $min_servers;
178     $max_spare ||= $max_servers;
179     $max_requests ||= 1000;
180
181     $logger->info("UnixServer: min=$min_servers, max=$max_servers, min_spare=$min_spare ".
182         "max_spare=$max_spare, max_req=$max_requests, log_file=$log_file, port=$port, pid_file=$pid_file");
183
184     $self->run(
185         min_servers => $min_servers,
186         max_servers => $max_servers,
187         min_spare_servers => $min_spare,
188         max_spare_servers => $max_spare,
189         max_requests => $max_requests,
190         log_file => $log_file,
191         port => $port,
192         proto => 'unix',
193         pid_file => $pid_file,
194     );
195
196 }
197
198
199 sub configure_hook {
200         my $self = shift;
201         my $app = $self->app;
202
203         # boot a client
204         OpenSRF::System->bootstrap_client( client_name => "system_client" );
205
206         $logger->debug( "Setting application implementaion for $app", DEBUG );
207         my $client = OpenSRF::Utils::SettingsClient->new();
208         my $imp = $client->config_value("apps", $app, "implementation");
209         OpenSRF::Application::server_class($app);
210         OpenSRF::Application->application_implementation( $imp );
211         JSON->register_class_hint( name => $imp, hint => $app, type => "hash" );
212         OpenSRF::Application->application_implementation->initialize()
213                 if (OpenSRF::Application->application_implementation->can('initialize'));
214
215         if( $client->config_value("server_type") !~ /fork/i  ) {
216                 $self->child_init_hook();
217         }
218
219         my $con = OpenSRF::Transport::PeerHandle->retrieve;
220         if($con) {
221                 $con->disconnect;
222         }
223
224         return OpenSRF::Application->application_implementation;
225 }
226
227 sub child_init_hook { 
228
229         $0 =~ s/master/drone/g;
230
231         if ($ENV{OPENSRF_PROFILE}) {
232                 my $file = $0;
233                 $file =~ s/\W/_/go;
234                 eval "use Devel::Profiler output_file => '/tmp/profiler_$file.out', buffer_size => 0;";
235                 if ($@) {
236                         $logger->debug("Could not load Devel::Profiler: $@",ERROR);
237                 } else {
238                         $0 .= ' [PROFILING]';
239                         $logger->debug("Running under Devel::Profiler", INFO);
240                 }
241         }
242
243         my $self = shift;
244
245 #       $logger->transport( 
246 #                       "Creating PeerHandle from UnixServer child_init_hook", INTERNAL );
247         OpenSRF::Transport::PeerHandle->construct( $self->app() );
248         $logger->transport( "PeerHandle Created from UnixServer child_init_hook", INTERNAL );
249
250         OpenSRF::Application->application_implementation->child_init
251                 if (OpenSRF::Application->application_implementation->can('child_init'));
252
253         return OpenSRF::Transport::PeerHandle->retrieve;
254 }
255
256 sub child_finish_hook {
257     $logger->debug("attempting to call child exit handler...");
258         OpenSRF::Application->application_implementation->child_exit
259                 if (OpenSRF::Application->application_implementation->can('child_exit'));
260 }
261
262
263 1;
264