gistfile1.perl

package Worker;
use Moose;
use Try::Tiny;
with qw(MooseX::Workers);
use POE qw(Filter::Reference Filter::Line);

sub run {
  my $self = shift;
  my @jobs = @{$self->get_jobs("jobs.txt")};
  for my $job ( @jobs ) {
    $self->enqueue(sub { $self->execute_job($job) } );
  }
  POE::Kernel->run();
}

# poll the job queue
sub get_jobs {
  my ( $self, $job_db ) = @_;
  open my $fh, "<", $job_db || die "Error opening file $job_db: $!";
  my @jobs = split "\n", do { local $/; <$fh> };
  print join ",", @jobs;
  return \@jobs;
}

# execute a job, this would normally be a dispatch table
# of some sort mapping classes to jobs
sub execute_job {
  my ( $self, $job ) = @_;
  try {
    $self->$job;
    print @{POE::Filter::Reference->new->put([ 
      {
        msg =>  "completed job successfully" 
      }
    ])};
  } catch {
    print @{POE::Filter::Reference->new->put([ 
      {
        msg =>   "Whoops, error running $job: $_"
      }
    ])};
  };
}

sub stdout_filter { POE::Filter::Reference->new }
sub stderr_filter { POE::Filter::Line->new }

sub worker_stdout {
  my ( $self, $result ) = @_;
  print $result->{msg} . "\n";

}

sub worker_stderr {
  my ( $self, $stderr_msg ) = @_;
  warn $stderr_msg;
}

sub worker_manager_start { warn 'started worker manager' }
sub worker_manager_stop  { warn 'stopped worker manager' }

sub max_workers_reached { warn 'maximum worker count reached' }
sub worker_error        { shift; warn join ' ', @_; }
sub worker_done         { shift; warn join ' ', @_; }
sub worker_started      { shift; warn join ' ', @_; }
sub sig_child           { shift; warn join ' ', @_; }
sub sig_TERM { shift; warn 'Handled TERM' }

sub job1 { 
  print @{POE::Filter::Reference->new->put([ {msg => "job 1 doing stuff" } ])}
}

sub job2 {
  print @{POE::Filter::Reference->new->put([ {msg => "job 2 doing stuff"} ])}
}
sub job3 {
  print @{POE::Filter::Reference->new->put([ {msg => "job 3 doing stuff"} ])}
}

no Moose;

jobs.txt

job1
job2
job3
job1
job3
job1
job1
job2
job3
job2
job2
job2
job2
job2
job2
job2
job3

worker.pl

use strictures;
use Worker;
my $worker = Worker->new;
$worker->run;

Worker.pm

package Worker;
use Moose;
use Try::Tiny;
with qw(MooseX::Workers);
use POE qw(Filter::Reference Filter::Line);

sub run {
  my $self = shift;
  my @jobs = @{$self->get_jobs("jobs.txt")};
  for my $job ( @jobs ) {
    $self->enqueue(sub { $self->execute_job($job) } );
  }
  POE::Kernel->run();
}

# poll the job queue
sub get_jobs {
  my ( $self, $job_db ) = @_;
  open my $fh, "<", $job_db || die "Error opening file $job_db: $!";
  my @jobs = split "\n", do { local $/; <$fh> };
  print join ",", @jobs;
  return \@jobs;
}

# execute a job, this would normally be a dispatch table
# of some sort mapping classes to jobs
sub execute_job {
  my ( $self, $job ) = @_;
  try {
    $self->$job;
    print @{POE::Filter::Reference->new->put([ 
      {
        msg =>  "completed job successfully" 
      }
    ])};
  } catch {
    print @{POE::Filter::Reference->new->put([ 
      {
        msg =>   "Whoops, error running $job: $_"
      }
    ])};
  };
}

sub stdout_filter { POE::Filter::Reference->new }
sub stderr_filter { POE::Filter::Line->new }

sub worker_stdout {
  my ( $self, $result ) = @_;
  print $result->{msg} . "\n";

}

sub worker_stderr {
  my ( $self, $stderr_msg ) = @_;
  warn $stderr_msg;
}

sub worker_manager_start { warn 'started worker manager' }
sub worker_manager_stop  { warn 'stopped worker manager' }

sub max_workers_reached { warn 'maximum worker count reached' }
sub worker_error        { shift; warn join ' ', @_; }
sub worker_done         { shift; warn join ' ', @_; }
sub worker_started      { shift; warn join ' ', @_; }
sub sig_child           { shift; warn join ' ', @_; }
sub sig_TERM { shift; warn 'Handled TERM' }

sub job1 { 
  print @{POE::Filter::Reference->new->put([ {msg => "job 1 doing stuff" } ])}
}

sub job2 {
  print @{POE::Filter::Reference->new->put([ {msg => "job 2 doing stuff"} ])}
}
sub job3 {
  print @{POE::Filter::Reference->new->put([ {msg => "job 3 doing stuff"} ])}
}

no Moose;