File Coverage

blib/lib/Proc/JobQueue/EventQueue.pm

Criterion	Covered	Total	%
statement	18	49	36.7
branch	0	12	0.0
condition	0	9	0.0
subroutine	6	11	54.5
pod	3	3	100.0
total	27	84	32.1

package Proc::JobQueue::EventQueue;

use strict;
use warnings;
use Carp qw(confess);
require Proc::JobQueue;
use Time::HiRes qw(time);
use Object::Dependency;
require POSIX;

our @ISA = qw(Proc::JobQueue);

our $timer_interval = 6;
my $debug = 0;

sub new
{
	my ($pkg, %params) = @_;

	$params{dependency_graph} ||= Object::Dependency->new();

	my $queue = $pkg->SUPER::new(
		unloop			=> undef,
		startmore_in_progress	=> 0,
		on_failure		=> \&on_failure,
		%params
	);

	my $last_dump = time;

	my $timer = IO::Event->timer(
		interval	=> $params{timer_interval} || $timer_interval,
		cb		=> sub {
			print STDERR "beep!\n" if $debug;
			eval {
				$queue->startmore;
			};
			if ($@) {
				print STDERR "DIE DIE DIE DIE DIE (DT1): $@";
				# exit 1; hangs
				POSIX::_exit(1);
			};
			if ($debug && time > $last_dump + $timer_interval) {
				$params{dependency_graph}->dump_graph();
				$last_dump = time;
			}
			use POSIX ":sys_wait_h";
			my $k;
			do { $k = waitpid(-1, WNOHANG) } while $k > 0;
		},
	);

	$Event::DIED = sub {
		Event::verbose_exception_handler(@_);
		$queue->unloop();
		IO::Event::unloop_all();
	};

	return $queue;
}

sub unloop
{
	my ($queue) = @_;
	if ($queue->{unloop}) {
		$queue->unloop($queue->alldone);
	} else {
		IO::Event::unloop_all();
	}
}

sub on_failure
{
	my ($queue, $job, @exit_code) = @_;
	if ($job->{on_failure}) {
		$job->{on_failure}->(@exit_code);
	} elsif ($job->{errors}) {
		$job->{errors}->("FAILED: $job->{desc}", @exit_code);
	} else {
		print STDERR "JOB $job->{desc} FAILED\nexit @exit_code\n";
	}
}


1;

__END__