File Coverage

blib/lib/Parallel/Queue/Manager.pm

Criterion	Covered	Total	%
statement	59	63	93.6
branch	14	18	77.7
condition	3	5	60.0
subroutine	18	18	100.0
pod	0	10	0.0
total	94	114	82.4

########################################################################
# housekeeping
########################################################################

package Parallel::Queue::Manager;
use v5.24;
use mro qw( c3 );

use Parallel::Queue qw( noexport );

use mro::EVERY;

use Carp            qw( croak           );
use Scalar::Util    qw( blessed reftype );

########################################################################
# package variables
########################################################################

our @CARP_NOT   = ( __PACKAGE__, qw( mro mro::EVERY ) );

my $parent_pid  = $$;

########################################################################
# methods
########################################################################

sub handler : lvalue
{
    my $qmgr    = shift;
    @_ or return $qmgr->[0];

    my $handler = shift;

    'CODE' eq reftype $handler
    or croak "handler: '$handler' is not CODE.";

    $qmgr->[0] = $handler
}

sub queue : lvalue
{
    my $qmgr    = shift;
    @_ or return $qmgr->[1] ||= [];

    $qmgr->[1]  = shift
}

sub next
{
    my $qmgr    = shift;
    my $queue   = $qmgr->queue;

    # the caller may want undef as an argument, who knows?
    # only fix is an exception to indicate no further jobs.

    @$queue
    ? shift @$queue
    : die "Empty queue.\n"
}

sub configure
{
    my $handler = Parallel::Queue->can( 'configure' );
    my $qmgr    = shift;

    $handler->( @_, qw( noexport ) );

    $qmgr
}

sub runqueue
{
    state $runq = Parallel::Queue->can( 'runqueue' );

    $parent_pid = $$;

    my $qmgr    = shift;
    my $jobs    = shift;

    # note that the queue may already be loaded
    # from construction or previous assignment.

    $qmgr->queue    = [ @_ ]
    if @_;

    $runq->( $jobs, $qmgr );
    $qmgr
}

sub next_job
{
    # pull the item off the stack in the
    # parent process, not the child. ignore
    # $@, eval returning undef is sufficient
    # to end queue execution.

    my $qmgr    = shift;

    eval
    {
        my $next    = $qmgr->next;
        sub { $qmgr->handler->( $next ) }
    }
}

########################################################################
# object manglement
########################################################################

sub new
{
    my $qmgr    = &construct;

    $qmgr->EVERY::LAST::initialize( @_ );
    $qmgr
}

sub construct
{
    my $proto   = shift;

    bless [], blessed $proto || $proto;
}

sub initialize
{
    my $qmgr        = shift;

    $qmgr->handler  = shift     if @_;
    $qmgr->queue    = [ @_ ]    if @_;

    return
}

DESTROY
{
    my $qmgr    = shift;

    $qmgr->EVERY::cleanup;

    undef @$qmgr;
    undef  $qmgr;

    return
}

sub cleanup
{
    if( $$ == $parent_pid )
    {
        my $qmgr    = shift;
        my $queue   = $qmgr->queue;

        say STDERR join "\n\t", "($$) Incomplete jobs:", @$queue
        if @$queue;
    }
    else
    {
        # child running individual job has nothing to
        # clean up.
    }

    return
}

# keep require happy
1
__END__