File Coverage

blib/lib/Parallel/Async/Chain.pm

Criterion	Covered	Total	%
statement	62	66	93.9
branch	5	6	83.3
condition			n/a
subroutine	16	17	94.1
pod	6	7	85.7
total	89	96	92.7

package Parallel::Async::Chain;
use 5.008005;
use strict;
use warnings;

use Class::Accessor::Lite rw => [qw/tasks/];
use POSIX ":sys_wait_h";
use Time::HiRes ();

sub join :method {
    my $self = shift;
    $self = bless +{ tasks => [] } => $self unless ref $self;

    push @{ $self->{tasks} } => @_;

    return $self;
}

sub recv :method {
    my ($self, @args) = @_;

    no warnings 'once';
    local $Parallel::Async::Task::WANTARRAY = 1;
    use warnings 'once';

    my @pids = $self->run(@args);
    $self->_wait(@pids);

    return $self->read_child_result();
}

sub run {
    my ($self, @args) = @_;
    return map { $_->run(@args) } @{ $self->{tasks} };
}

sub daemonize {
    my ($self, @args) = @_;
    return map { $_->daemonize(@args) } @{ $self->{tasks} };
}

sub _wait {
    my $self = shift;
    my %pids = map { $_ => 1 } @_;

    while (%pids) {
        my $pid = waitpid(-1, WNOHANG);
        last if $pid == -1;

        delete $pids{$pid} if exists $pids{$pid};
    }
    continue {
        no warnings 'once';
        Time::HiRes::usleep($Parallel::Async::Task::WAIT_INTERVAL);
    }
}

sub read_child_result {
    my $self = shift;
    return map { $_->read_child_result() } @{ $self->{tasks} };
}

sub reset :method {
    my $self = shift;
    $_->reset for @{ $self->{tasks} };
    return $self;
}

sub clone {
    my $self  = shift;
    my $class = ref $self;
    return $class->join(map { $_->clone } @{ $self->{tasks} });
}

1;
__END__