a simple worker; a role for the REST interface to presque; reg signals

to shutdown workers; log before starting a task; fork dispatcher (a la resque)

3 files changed, 46 insertions, 8 deletions

diff --git a/lib/presque/worker/Role/Fork.pm b/lib/presque/worker/Role/Fork.pm
index 47efc31..c62ff1b 100644
--- a/lib/presque/worker/Role/Fork.pm
+++ b/lib/presque/worker/Role/Fork.pm
@@ -5,8 +5,7 @@ use Moose::Role;
 has fork_dispatcher => (
     is        => 'ro',
     isa       => 'Bool',
-    default   => 1,
-    predicate => 'has_fork_dispatcher'
+    default   => 0,
 );
 
 1;
diff --git a/lib/presque/worker/Role/Management.pm b/lib/presque/worker/Role/Management.pm
index 2bd4db3..cdfcd6d 100644
--- a/lib/presque/worker/Role/Management.pm
+++ b/lib/presque/worker/Role/Management.pm
@@ -6,7 +6,7 @@ has shut_down => (is => 'rw', isa => 'Bool', default => 0,);
 
 before start => sub {
     my $self = shift;
-    $self->rest_register_worker
+    $self->rest_register_worker;
 };
 
 after start => sub {
@@ -14,6 +14,41 @@ after start => sub {
     $self->rest_unregister_worker;
 };
 
-# XXX reg signal
+before start => sub {
+    my $self = shift;
+    $SIG{'INT'}  = sub { $self->_shutdown };
+    $SIG{'TERM'} = sub { $self->_shutdown };
+    $SIG{'QUIT'} = sub { $self->_graceful_shutdown };
+    $SIG{'USR1'} = sub { $self->_kill_child };
+};
+
+sub _shutdown {
+    my $self = shift;
+    $self->logger->log(
+        level   => 'info',
+        message => 'worker ' . $self->worker_id . ' shuting down'
+    );
+    $self->shut_down(1);
+    $self->_kill_child();
+}
+
+sub _graceful_shutdown {
+    my $self = shift;
+    $self->logger->log(
+        level   => 'info',
+        message => 'worker ' . $self->worker_id . ' kill child'
+    );
+    $self->shut_down(1);
+    $self->_kill_child();
+}
+
+sub _kill_child {
+    my $self = shift;
+    $self->logger->log(
+        level   => 'info',
+        message => 'worker ' . $self->worker_id . ' shuting down gracefuly'
+    );
+}
 
 1;
+
diff --git a/lib/presque/worker/Role/RESTClient.pm b/lib/presque/worker/Role/RESTClient.pm
index 0015b98..dd84fda 100644
--- a/lib/presque/worker/Role/RESTClient.pm
+++ b/lib/presque/worker/Role/RESTClient.pm
@@ -39,13 +39,14 @@ sub rest_register_worker {
 
 sub rest_unregister_worker {
     my $self = shift;
-    my $request = HTTP::Request->new(DELETE => $self->_worker_uri);
-    $request->query_path(worker_id => $self->worker_id);
+    my $uri  = $self->_worker_uri;
+    $uri->query_form(worker_id => $self->worker_id);
+    my $request = HTTP::Request->new(DELETE => $uri);
     my $res = $self->ua->request($request);
 }
 
 sub rest_fetch_job {
-    my ($self,) = @_;
+    my $self = shift;
 
     my $res = $self->ua->request(HTTP::Request->new(GET => $self->_job_uri));
     if ($res->is_success) {
@@ -57,20 +58,23 @@ sub rest_fetch_job {
             message => $res->code . ':' . $res->message
         );
     }
+    return;
 }
 
 sub rest_retry_job {
     my ($self, $job) = @_;
 
     my $request = HTTP::Request->new(PUT => $self->_job_uri);
+    $request->header('Content-Type' => 'application/json');
     $request->content(JSON::encode_json($job));
     my $res = $self->ua->request($request);
     if (!$res->is_success) {
+        use YAML::Syck; warn Dump $res;
         $self->logger->log(
             level   => 'error',
             message => 'failed to update job ('
               . $res->code . ':'
-              . $res->reason . ')',
+              . $res->message . ')',
         );
     }
 }