POE

这两天比较空，刚看点了POE ，于是就用POE写了一个daemon，和crontab的作用差不多。主要是自己写的，控制方面更灵活一些。现在提供了基本的start，stop，restart，state方法。要加新的任务只要在指定目录中添加相应的模块就可以了。用起来还算是简单吧。不过POE是单进程的，不知道如果挂的任务多了会不会太慢。现在虽然写出了个框，以后却不一定用的上，以后弄服务器争取能用一下试试。

代码：

  1  #!/usr/bin/perl
  2
  3
  4  # Last Modified: <2006-02-21 16:37:57>
  5
  6  use strict;
  7  use POE qw{
  8    Component::Server::TCP
  9    Component::Client::TCP
 10  };
 11  use FindBin    ();
 12  use File::Find ();
 13  use Data::Dumper;
 14
 15  use constant PORT        => 60000;
 16  use constant MODULE_PATH => '';      # need absolute path
 17
 18  our $VERSION = 1.1;
 19  our %Sessions;
 20
 21  main();
 22
 23  sub main {
 24
 25      #become_daemon();
 26      my $cmd = $ARGV[0];
 27      if ($cmd) {
 28          if ($cmd eq 'start') {
 29              start();
 30          } else {
 31              client($cmd);
 32          }
 33      } else {
 34          usage();
 35      }
 36      POE::Kernel->run();
 37      exit(0);
 38  }
 39
 40  sub usage {
 41      print "Usage: $0 start|stop|state|restart\n";
 42  }
 43
 44  # daemon server
 45
 46  sub start {
 47
 48      # start daemon
 49      POE::Session->create(
 50          inline_states => {
 51              start         => \&load_module,
 52              stop          => \&stop,
 53              restart       => \&restart,
 54              state         => \&state,
 55              check_session => \&check_session,
 56              _start => sub {
 57                  $_[KERNEL]->alias_set('daemon');
 58                  $_[KERNEL]->yield('start');
 59              },
 60          }
 61      );
 62
 63      POE::Component::Server::TCP->new(
 64          Alias              => 'daemon_server',
 65          Address            => "localhost",
 66          Port               => PORT,
 67          ClientInput        => \&server_command,
 68          ClientDisconnected => sub {
 69              $_[KERNEL]->post(daemon_server => 'shutdown') if $_[HEAP]->{stop_server} == 2;
 70          },
 71          InlineStates => { response => \&response, },
 72      );
 73  }
 74
 75  sub stop {
 76      my ($kernel, $heap, $sender) = @_[ KERNEL, HEAP, ARG0 ];
 77      if ($heap->{stop_client_handler}) {
 78          $kernel->post($heap->{stop_client_handler}, 'response', 'OK:stop')
 79            if ref($heap->{stop_client_handler});
 80          delete $heap->{stop_client_handler};
 81      } else {
 82          $heap->{stop_client_handler} = $sender || 1;
 83          stop_module($kernel);
 84          $heap->{check_result} = 'stop';
 85          $kernel->yield('check_session');
 86      }
 87  }
 88
 89  sub state {
 90      my ($kernel, $heap, $sender) = @_[ KERNEL, HEAP, SENDER ];
 91      my $msg;
 92      for my $s (keys %Sessions) {
 93          $msg = $kernel->call($s => 'state_session');
 94          $msg = "Has problem, $!" if $!;
 95          $kernel->call($sender, 'response', "$s : $msg");
 96      }
 97      $kernel->post($sender, 'response', "OK:Done");
 98  }
 99
100  sub restart {
101      my ($kernel, $heap, $sender) =
102        @_[ KERNEL, HEAP, ARG0 ];    # sender may shutdown, so do not use SENDER
103      if ($heap->{restart_client_handler}) {
104          load_module();
105          $kernel->post($heap->{restart_client_handler}, 'response', 'OK:restart')
106            if ref($heap->{restart_client_handler});
107          delete $heap->{restart_client_handler};
108      } else {
109          $heap->{restart_client_handler} = $sender || 1;
110          stop_module($kernel);
111          $heap->{check_result} = 'restart';
112          $kernel->yield('check_session');
113      }
114  }
115
116  sub check_session {
117
118      # check if session is stopped
119      my ($kernel, $heap) = @_[ KERNEL, HEAP ];
120      if (%Sessions) {
121          $kernel->delay_add(check_session => 0.5);
122      } else {
123          $kernel->yield($heap->{check_result});
124      }
125  }
126
127  sub load_module {
128      my $secord  = 0;
129      my @modules = find_module();
130      for my $p (@modules) {
131          eval "require $p";
132          next if $@;
133          {
134              no strict 'refs';
135              *{"${p}::_stop"} = sub {
136                  delete $Sessions{$p};
137                  print "stop session ", $_[KERNEL]->alias_list, "\n";
138              };
139              *{"${p}::_start"} = sub {
140                  my $k = $_[KERNEL];
141                  $k->alias_set($p);
142                  $k->sig(TERM => 'stop_session');
143                  $k->delay_add('start_session', $secord);
144              };
145              unless (UNIVERSAL::can($p, 'stop_session')) {
146                  *{"${p}::stop_session"} = sub { return 0; };
147              }
148              unless (UNIVERSAL::can($p, 'state_session')) {
149                  *{"${p}::state_session"} = sub { return 'OK'; };
150              }
151          }
152
153          my @methods = find_method($p);
154          next unless @methods;
155          $Sessions{$p} = 1;
156          POE::Session->create(package_states => [ $p => [@methods] ]);
157          $secord++;
158          $secord = 0 if $secord > 60;
159      }
160  }
161
162  sub stop_module {
163      my $kernel = $_[0];
164      for my $s (keys %Sessions) {
165          $kernel->signal($s, 'TERM');
166      }
167  }
168
169  # client
170
171  sub client {
172      my $cmd = shift;
173      POE::Component::Client::TCP->new(
174          Alias         => "daemon_client",
175          RemoteAddress => "localhost",
176          RemotePort    => PORT,
177          Connected     => sub {
178              $_[HEAP]->{server}->put("$cmd");
179          },
180          ServerInput => sub {
181              my ($kernel, $res) = @_[ KERNEL, ARG0 ];
182              local $| = 1;
183              print $res, "\n";
184              if ($res =~ m/^(OK|Error):(.*)$/) {
185                  $kernel->yield('shutdown');
186              }
187          },
188      );
189  }
190
191  sub server_command {
192      my ($kernel, $heap, $cmd) = @_[ KERNEL, HEAP, ARG0 ];
193      $heap->{stop_server} = 0;
194      if ($cmd eq 'stop') {
195          $heap->{client}->put("Stopping...");
196          $heap->{stop_server} = 1;
197          $kernel->post(daemon => 'stop', $_[SESSION]);
198      } elsif ($cmd eq 'state') {
199          $heap->{client}->put("Checking...");
200          $kernel->post(daemon => 'state');
201      } elsif ($cmd eq 'restart') {
202          $heap->{client}->put('Restart...');
203          $kernel->post(daemon => 'restart', $_[SESSION]);
204      } else {
205          $kernel->yield(response => "Error: Bad command");
206      }
207  }
208
209  sub response {
210      my $heap = $_[HEAP];
211      $heap->{stop_server} = 2 if $heap->{stop_server};
212      $heap->{client}->put(join "\n", @_[ ARG0 .. $#_ ]);
213  }
214
215  # other
216
217  sub find_method {
218      my $package = shift;
219      my @methods;
220      no strict 'refs';
221      for my $k (keys %{"${package}::"}) {
222          if ($k !~ m/^__/ && UNIVERSAL::can($package, $k)) {
223              push @methods, $k;
224          }
225      }
226      return @methods;
227  }
228
229  sub become_daemon {
230      my $child = fork;
231      die "Can't fork: $!" unless defined($child);
232      exit(0) if $child;    # parent dies;
233      POSIX::setsid();      # become session leader
234      open(STDIN,  "</dev/null");
235      open(STDOUT, ">/dev/null");
236      open(STDERR, '>&STDOUT');
237      umask(0);             # forget file mode creation mask
238      $ENV{PATH} = '/bin:/sbin:/usr/bin:/usr/sbin:/usr/local/bin';
239      delete @ENV{ 'IFS', 'CDPATH', 'ENV', 'BASH_ENV' };
240  }
241
242  sub find_module {
243      my $dir   = $FindBin::Bin;
244      my $regex = '\.pm$';
245      my @list;
246      my $pre_dir = $dir;
247      $pre_dir =~ s{[\\\/]}{::}g;
#cc

  


  

    
    
分享到：
      
      
    
  

  

    

      free book
      |
      emacs duplicate line
    
  
  

    

      
2006-02-21 12:51
      
浏览 618
      
评论(0)
      
      
            
      
查看更多