fork_pmap.pl

package MyPmap::Filter;
use metaclass 'MooseX::POE::SweetArgs';
use MooseX::POE;
use POE::Wheel::Run;

has cont => (
	isa => 'ArrayRef',
	is  => 'ro',
	required => 1,
);

has filter => (
	isa => 'CodeRef',
	is  => 'ro',
	required => 1,
);

has child => (
	isa => 'Maybe[POE::Wheel::Run]',
	is  => 'rw',
);

sub STOP{
	warn __PACKAGE__ . " done\n";
}

event exec => sub {
	my $self = shift;
	my ($val) = @_;
	my $filter = $self->filter;  # for closure
	my $task = POE::Wheel::Run->new(
		Program => sub { print $filter->($val), "\n"; },
		StdoutEvent => "stdout",
		StderrEvent => "stderr",
		CloseEvent  => "closed",
	);

	$self->child( $task );
	$POE::Kernel::poe_kernel->sig_child( $task->PID, "reaped" );

};

event stdout => sub {
	my $self = shift;
	my ($val) = @_;
	$POE::Kernel::poe_kernel->post(
		@{ $self->cont } =>
		$val
	);
};

event stderr => sub{ warn 'ERROR:' , @_ };

event closed => sub{ 
	my $self = shift;
	$self->child( undef );
};

event reaped => sub{ 
	# NOP
	# warn 'REAPED:', @_
};

no MooseX::POE;


package MyPmap;
use metaclass 'MooseX::POE::SweetArgs';
use MooseX::POE;
use POE;

has cont => (
	isa => 'ArrayRef',
	is  => 'ro',
	required => 1,
);

has filter => (
	isa => 'CodeRef',
	is  => 'ro',
	required => 1,
);

has count => (
	isa => 'Int',
	is  => 'rw',
);

has result => (
	isa     => 'ArrayRef',
	is      => 'rw',
	default => sub { [] },
);

sub STOP{
	warn __PACKAGE__ . " done\n";
}

event exec => sub {
	my $self = shift;
	my ( @args ) = @_;

	$self->count(scalar @args);

	for(@args){
		MyPmap::Filter->new(
			cont   => [$self->get_session_id => 'recieve'],
			filter => $self->filter,
		)->yield('exec' => $_);
	}
};

event recieve => sub {
	my $self = shift;
	my ($ret) = @_;

	$self->count($self->count - 1);
	push @{ $self->result }, $ret;

	if($self->count == 0){
		$POE::Kernel::poe_kernel->post(
			@{ $self->cont } =>
			@{ $self->result }
		);
	}
};

no MooseX::POE;


package MySession;
use metaclass 'MooseX::POE::SweetArgs';
use MooseX::POE;

sub START{
	my $self = shift;
	MyPmap->new(
		cont   => [$self->get_session_id, 'recieve'],
		filter => sub { sleep 1; $_[0] * 2 },
	)->yield('exec', 1 .. 10);
}

sub STOP{
	warn __PACKAGE__ . " done\n";
}

event recieve => sub {
	my $self = shift;
	my @odds = @_;
	print join(",", @odds), "\n";
	print "done\n";
};

no MooseX::POE;


package main;
use strict;
use warnings;
use POE;
MySession->new;
POE::Kernel->run;