File Coverage

blib/lib/AnyEvent/ProcessPool/Pipeline.pm

Criterion	Covered	Total	%
statement	31	31	100.0
branch			n/a
condition	1	3	33.3
subroutine	9	9	100.0
pod	3	3	100.0
total	44	46	95.6

package AnyEvent::ProcessPool::Pipeline;
# ABSTRACT: A simplified, straightforward way to parallelize tasks
$AnyEvent::ProcessPool::Pipeline::VERSION = '0.06';
use strict;
use warnings;
use AnyEvent::ProcessPool;
use Try::Catch;

use parent 'Exporter';

our @EXPORT = qw(pipeline in out);

sub pipeline (%) {
  my %param = @_;
  my $in    = delete $param{in};
  my $out   = delete $param{out};
  my $pool  = delete $param{pool} || AnyEvent::ProcessPool->new(%param);
  my $count = 0;

  my %pending;
  while (my @task = $in->()) {
    my $cv = $pool->async(@task);
    $pending{$cv} = $cv;
    $cv->cb(sub{ ++$count; $out->(shift) });
  }

  $pool->join; # wait for all tasks to complete

  return $count;
}

sub in  (&) { return (in  => $_[0]) }
sub out (&) { return (out => $_[0]) }

1;

__END__