File Coverage

blib/lib/Parallel/Map/Segmented.pm

Criterion	Covered	Total	%
statement	26	30	86.6
branch	5	8	62.5
condition	2	3	66.6
subroutine	7	8	87.5
pod	1	1	100.0
total	41	50	82.0

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

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
    {
        while (
            defined(
                $batch = $stream_cb->( { size => $batch_size } )->{items}
            )
            )
        {
            $batch_cb->($batch);
        }
    }
    return;
}

1;

__END__