File Coverage

blib/lib/Job/Machine/Worker.pm

Criterion	Covered	Total	%
statement	9	68	13.2
branch	0	4	0.0
condition	0	20	0.0
subroutine	3	21	14.2
pod	12	12	100.0
total	24	125	19.2

package Job::Machine::Worker;
$Job::Machine::Worker::VERSION = '0.26';
use strict;
use warnings;

use base 'Job::Machine::Base';

sub reply {
	my ($self,$data,$queue) = @_;
	my $db = $self->db;
	$queue ||= $self->{queue};
	$self->result($data,$queue);
	my $task_id = $db->task_id;
## Payload: Status of result, result id...
	$db->notify(queue => $task_id, reply => 1);
	return $task_id;
}

sub result {
	my ($self,$data,$queue) = @_;
	$queue ||= $self->{queue};
	$self->db->insert_result($data,$queue);
	$self->db->set_task_status(200);
}

sub error_result {
	my ($self,$data,$queue) = @_;
	$queue ||= $self->{queue};
	$self->db->insert_result($data,$queue);
}

sub receive {
	my $self = shift;
	$self->startup;
	my $db = $self->{db};
	$self->_init_chores;
	$self->subscribe($self->{queue});
	$self->_check_queue($self->{queue});
	while ($self->keep_running && (my $notifies = $db->set_listen($self->timeout))) {
		my ($queue,$pid) = @$notifies;
		$self->_do_chores() && next unless $queue;

		$self->_check_queue($self->{queue});
	}
	return;
};

sub _check_queue {
	my $self = shift;
	my $db = $self->{db};
	while (my $task = $self->db->fetch_work_task) {
		## log process call
		$self->process($task);
	}
}


sub _init_chores {
	my $self = shift;
	my $db = $self->{db};
	my @chores = (
		sub {
			my $self = shift;
			my $number = $db->revive_tasks($self->max_runtime) || 0;
			$self->job_log("Revived tasks: $number");
		},
		sub {
			my $self = shift;
			my $number = $db->fail_tasks($self->retries) || 0;
			$self->job_log("Failed tasks: $number");
		},
		sub {
			my $self = shift;
			my $number = $db->remove_tasks($self->remove_after) || 0;
			$self->job_log("Removed tasks: $number");
		},
	);
	push @{ $self->{chores} }, @chores;
}


sub add_chore {
	my ($self, $chore) = @_ ;
	return unless ref $chore eq 'CODE';

	push @{ $self->{chores} }, $chore;
}

sub _do_chores {
	my $self = shift;
	my $chores = $self->{chores};
	my $idx = int(rand(@{ $chores }));
	my $chore = $chores->[$idx];
	$self->$chore;
}

sub startup {}

sub process {die 'Subclasss me!'}

sub max_runtime {return 30*60}

sub timeout {return 300}

sub retries {return 3}

sub remove_after {return 30}

sub keep_running {return 1}

1;

__END__