File Coverage

blib/lib/Parallel/Map/Segmented.pm

Criterion	Covered	Total	%
statement	32	35	91.4
branch	5	8	62.5
condition	2	3	66.6
subroutine	9	10	90.0
pod	1	1	100.0
total	49	57	85.9

package Parallel::Map::Segmented;
$Parallel::Map::Segmented::VERSION = '0.4.1';
use strict;
use warnings;
use autodie;
use 5.014;

use Parallel::ForkManager::Segmented::Base v0.4.0;
use parent 'Parallel::ForkManager::Segmented::Base';

use Parallel::Map qw/ pmap_void /;

sub run
{
    my ( $self, $args ) = @_;

    my $processed = $self->process_args($args);
    return if not $processed;
    my ( $WITH_PM, $batch_cb, $batch_size, $nproc, $stream_cb, ) =
        @{$processed}{qw/ WITH_PM batch_cb batch_size nproc stream_cb  /};

    my $batch = $stream_cb->( { size => 1 } )->{items};
    return if not defined $batch;
    $batch_cb->($batch);
    if ($WITH_PM)
    {
        pmap_void sub { $batch_cb->(shift); Future->done; }, generate => sub {
            return $stream_cb->( { size => $batch_size } )->{items} // ();
            },
            ( $nproc ? ( forks => $nproc ) : () ),
            ;
    }
    else
    {
        $self->serial_run($processed);
    }
    return;
}

1;

__END__