135/netdisco
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

remove unnecessary Worker::Common role

Browse Source
This commit is contained in:
Oliver Gorwits
2017-09-03 16:22:29 +01:00
parent 36b4adcc06
commit 3baa7a818a
2 changed files with 61 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files

68
lib/App/Netdisco/Backend/Worker/Common.pm
View File

@@ -1,68 +0,0 @@
package App::Netdisco::Backend::Worker::Common;
use Dancer qw/:moose :syntax :script/;
use Try::Tiny;
use App::Netdisco::Util::MCE;
use Role::Tiny;
use namespace::clean;
use Time::HiRes 'sleep';
use App::Netdisco::JobQueue qw/jq_defer jq_complete/;
sub worker_begin { (shift)->{started} = time }
sub worker_body {
my $self = shift;
my $wid = $self->wid;
while (1) {
prctl sprintf 'nd2: #%s poll: idle', $wid;
my $job = $self->{queue}->dequeue(1);
next unless defined $job;
my $action = $job->action;
try {
$job->started(scalar localtime);
prctl sprintf 'nd2: #%s poll: #%s: %s',
$wid, $job->job, $job->summary;
info sprintf "pol (%s): starting %s job(%s) at %s",
$wid, $action, $job->job, $job->started;
my ($status, $log) = $self->$action($job);
$job->status($status);
$job->log($log);
}
catch {
$job->status('error');
$job->log("error running job: $_");
$self->sendto('stderr', $job->log ."\n");
};
$self->close_job($job);
sleep( setting('workers')->{'min_runtime'} || 0 );
$self->exit(0); # recycle worker
}
}
sub close_job {
my ($self, $job) = @_;
my $now = scalar localtime;
info sprintf "pol (%s): wrapping up %s job(%s) - status %s at %s",
$self->wid, $job->action, $job->job, $job->status, $now;
try {
if ($job->status eq 'defer') {
jq_defer($job);
}
else {
$job->finished($now);
jq_complete($job);
}
}
catch { $self->sendto('stderr', "error closing job: $_\n") };
}
1;

63
lib/App/Netdisco/Backend/Worker/Poller.pm
View File

@@ -1,10 +1,15 @@
package App::Netdisco::Backend::Worker::Poller; package App::Netdisco::Backend::Worker::Poller;
use Dancer qw/:moose :syntax :script/;
use Try::Tiny;
use App::Netdisco::Util::MCE;
use Role::Tiny; use Role::Tiny;
use namespace::clean; use namespace::clean;
# main worker body use Time::HiRes 'sleep';
with 'App::Netdisco::Backend::Worker::Common'; use App::Netdisco::JobQueue qw/jq_defer jq_complete/;
# add dispatch methods for poller tasks # add dispatch methods for poller tasks
with 'App::Netdisco::Backend::Worker::Poller::Device', with 'App::Netdisco::Backend::Worker::Poller::Device',
@@ -15,4 +20,58 @@ with 'App::Netdisco::Backend::Worker::Poller::Device',
'App::Netdisco::Backend::Worker::Interactive::DeviceActions', 'App::Netdisco::Backend::Worker::Interactive::DeviceActions',
'App::Netdisco::Backend::Worker::Interactive::PortActions'; 'App::Netdisco::Backend::Worker::Interactive::PortActions';
sub worker_begin { (shift)->{started} = time }
sub worker_body {
my $self = shift;
my $wid = $self->wid;
while (1) {
prctl sprintf 'nd2: #%s poll: idle', $wid;
my $job = $self->{queue}->dequeue(1);
next unless defined $job;
my $action = $job->action;
try {
$job->started(scalar localtime);
prctl sprintf 'nd2: #%s poll: #%s: %s',
$wid, $job->job, $job->summary;
info sprintf "pol (%s): starting %s job(%s) at %s",
$wid, $action, $job->job, $job->started;
my ($status, $log) = $self->$action($job);
$job->status($status);
$job->log($log);
}
catch {
$job->status('error');
$job->log("error running job: $_");
$self->sendto('stderr', $job->log ."\n");
};
$self->close_job($job);
sleep( setting('workers')->{'min_runtime'} || 0 );
$self->exit(0); # recycle worker
}
}
sub close_job {
my ($self, $job) = @_;
my $now = scalar localtime;
info sprintf "pol (%s): wrapping up %s job(%s) - status %s at %s",
$self->wid, $job->action, $job->job, $job->status, $now;
try {
if ($job->status eq 'defer') {
jq_defer($job);
}
else {
$job->finished($now);
jq_complete($job);
}
}
catch { $self->sendto('stderr', "error closing job: $_\n") };
}
1; 1;