File Coverage

blib/lib/Deeme/Worker.pm

Criterion	Covered	Total	%
statement	40	40	100.0
branch	7	10	70.0
condition	2	4	50.0
subroutine	8	8	100.0
pod	6	6	100.0
total	63	68	92.6

package Deeme::Worker;
use Deeme::Obj "Deeme";
use constant DEBUG => $ENV{DEEME_DEBUG} || 0;

sub jobs {
    my $j;
    return ( $j = shift->backend->events_get(shift) ) ? scalar @$j : 0;
}

sub dequeue_event {
    my ( $self, $name ) = @_;
    if ( my $s = $self->backend->events_get($name) ) {
        warn
            "Worker -- dequeue $name in @{[blessed $self]} (@{[scalar @$s]})\n"
            if DEBUG;
        my @onces = $self->backend->events_onces($name);
        my $i     = 0;
        for my $cb (@$s) {
            ( $onces[$i] == 1 )
                ? ( splice( @onces, $i, 1 )
                    and $self->_unsubscribe_index( $name => $i ) )
                : $i++;
            push(
                @{ $self->{'queue'} },
                Deeme::Job->new(
                    deeme => $self,
                    cb    => $cb
                )
            );
        }
    }
    return @{ $self->{'queue'} };
}

sub dequeue {
    my ( $self, $name ) = @_;

    if ( my $s = $self->backend->events_get($name) ) {
        warn
            "Worker -- dequeue $name in @{[blessed $self]} safely (@{[scalar @$s]})\n"
            if DEBUG;
        my $cb = Deeme::Job->new(
            deeme => $self,
            cb    => @$s[0]
        );
        $self->_unsubscribe_index( $name, 0 );
        push( @{ $self->{'queue'} }, $cb );
    }
    return @{ $self->{'queue'} }[0];
}

sub process {
    my $self = shift;
    return @{ $self->{'queue'} } > 0
        ? @{ $self->{'queue'} }[0]->process(@_)
        : undef;
}

sub process_all {
    my $self = shift;
    my @args = @_;
    my @returns;
    while ( my $job = shift @{ $self->{'queue'} } ) {
        push( @returns, $job->process(@args) );
    }
    return @returns;
}

sub add { return shift->once(@_) }
1;
__END__