Parent - child IPC through pipes

pipe_ipc.pl

#/usr/bin/perl
use strict;
use warnings;

use AnyEvent;
use AnyEvent::Handle;
use Parallel::SubFork;

my $manager = Parallel::SubFork->new();

my ($pipe_r, $pipe_w);

# Start two parallel tasks
my $closure = sub {
	## pipe_w is a file handle after pipe command is executed
	## we create pipe_w AnyEvent handle based on file handle
	$pipe_w = AnyEvent::Handle->new(fh => $pipe_w);

	## close pipe_r since we are not reading from child
	close $pipe_r;

	for (1..10) {
		sleep 1;
	    $pipe_w->push_write( json => {value => sprintf "Child PID $$: %d\n", $_} );
	}

	## pipe_w is closed implicitly
	print "Child $$ exiting.\n";
};

## Parent code
pipe $pipe_r, $pipe_w;

my $task = $manager->start($closure);

## we don't write from parent, just as we don't read from child
close $pipe_w;

## Condvar is primary point of synchronization that starts eventloop
my $cv = AnyEvent->condvar;

$pipe_r = AnyEvent::Handle->new(fh => $pipe_r);
$pipe_r->on_eof(sub {
	print "EOF reached!\n\n";
	undef $pipe_r;
	$cv->send();
});

$pipe_r->on_read( sub {
	## only push_read will make us reach EOF
	## on_read is PEEK, opush_read is POLL
	$pipe_r->push_read( json => sub {
		my $hdl  = shift;
        my $json = shift;

        print $json->{value}."\n";
	});
});

$cv->recv();
# Wait for all tasks to resume is not needed
# since we guarantee that EOF is reached with cv
# $manager->wait_for_all();