POE: Cookbook - Child Processes Nested

#!/usr/pin/perl
# This example demonstrates setting up a child process based on a coderef
# using POE::Wheel::Run and installing a nested kernel inside of it which talks
# back and forth with the parent process via POE::Filter::Reference.
#
# The worker is comically simple, it just increments a counter.
#
# It took me some time to figure this out so I hope others find it useful.
# It could probably be simplified somewhat.
use strict;
use warnings;
use POE qw(Wheel::Run Wheel::ReadWrite Filter::Reference);

sub worker {
  POE::Kernel->stop();
  print STDERR "Worker forked\n";
  POE::Session->create(
    inline_states => {
      _start => sub {
        my ($kernel, $heap) = @_[KERNEL, HEAP];
        $heap->{client} = POE::Wheel::ReadWrite->new(
          InputHandle  => \*STDIN,
          OutputHandle => \*STDOUT,
          InputEvent   => 'worker_inside_input',
          ErrorEvent   => 'worker_inside_error',
          Filter       => POE::Filter::Reference->new(),
        );
        $kernel->sig(TERM => '_stop');
        print STDERR "Worker session started\n";
      },
      worker_inside_input => sub {
        my ($heap, $kernel, $input) = @_[HEAP, KERNEL, ARG0];
        print STDERR "got input $input->[0]\n";
        my $output = [$input->[0] + 1];
        $heap->{client}->put($output);
      },
      worker_inside_error => sub {
        my ($heap, $op, $code) = @_[HEAP, ARG0, ARG1];
        warn "worker inside error: $op $code";
      },
    }
  ) or die "worker: can't POE::Session->create: $!";
  POE::Kernel->run();
  print STDERR "Worker done\n";
  return;
}

sub worker_error {
  my ($hash, $op, $code, $handle) = @_[HEAP, ARG0, ARG1, ARG4];
  if ($op eq 'read' and $code == 0 and $handle eq 'STDOUT') {
    warn "child has closed output";
    delete $hash->{worker};
  }
}

sub worker_stdout {
  my ($heap, $kernel, $value) = @_[HEAP, KERNEL, ARG0];
  $heap->{count} = $value->[0];
  print "worker responded: " . $heap->{count} . "\n";
  $kernel->yield('next_cmd');
}

sub worker_stderr {
  my ($heap, $txt) = @_[HEAP, ARG0];
  print "worker_stderr: $txt\n";
}
POE::Session->create(
  inline_states => {
    _start => sub {
      my ($kernel, $heap) = @_[KERNEL, HEAP];
      $heap->{worker} = POE::Wheel::Run->new(
        Program     => \&worker,
        ErrorEvent  => 'worker_error',
        StdoutEvent => 'worker_stdout',
        StderrEvent => 'worker_stderr',
        StdioFilter => POE::Filter::Reference->new(),
      ) or die "$0: can't POE::Wheel::Run->new";
      $heap->{count} = 0;
      $kernel->yield('next_cmd');
    },
    worker_stdout => \&worker_stdout,
    worker_stderr => \&worker_stderr,
    worker_error  => \&worker_error,
    next_cmd      => sub {
      my ($kernel, $heap) = @_[KERNEL, HEAP];
      if ($heap->{count} > 3) {
        $heap->{worker}->kill('TERM');
      }
      else {
        my $input = $heap->{count};
        $heap->{worker}->put([$input]);
      }
    },
  }
) or die "$0: POE::Session->create failed $!";
POE::Kernel->run();
exit 0;