I am writing a service, obtaining file names from an external source (other application is assumed to upload a file and send its name to my service in the HTTP GET request) and feeding the files to the processing engine. This engine sometimes can become busy. I would like to repeat the processing instead of dropping the file.

I use POE::Component::Server::HTTPServer to serve HTTP requests. Its handlers post messages with file names to passive POE::Component::JobQueue. The sessions, created by the workers of the queue, use POE::Wheel::Run::Win32, run another executable and capture its output.

The question is - where should I put the call to post event, when that executable writes "ENGINE IS BUSY"?

The code below works only first time.
It doesn't stop a created worker session - I don't see the messages, printed by the subroutine, assigned to _stop (scroll down to sub make_a_worker).
If I comment the call to POE::Kernel->post in the sub worker_closed, the session is destroyed, and the service is ready to process next file. 

use POE qw( Wheel::Run::Win32 Filter::Line);
use POE::Component::Server::HTTPServer;
use POE::Component::Server::HTTPServer::Handler;
use POE::Component::JobQueue;
use POE::Filter::Reference;
use File::Basename;
use Data::Dump qw(dump);
use Cwd;

my $port=8080;
my $max_jobs=1;

sub _start {
    my ( $kernel, $heap ) = @_[ KERNEL, HEAP ];

    $heap->{httpserver} = POE::Component::Server::HTTPServer->new(
        port=>$port,
        handlers => [ "/add" => \&add_task]
    );
    $heap->{httpserver}->create_server();
    
    POE::Component::JobQueue->spawn
        ( Alias         => 'passive',         
          WorkerLimit   => $max_jobs,  
          Worker        => \&make_a_worker,
          Passive       => { }
      );
}

sub add_task {
    my $context = shift;

# request has a form http://server:port/add?file/name.mpg
    $context->{request}->uri=~/\/add\?(.*)/;     my @job_params=($1);

    POE::Kernel->post( 'passive', 
                       'enqueue',   
                       'postback', 
                        @job_params
               );
    $context->{response}->code(200);
    $context->{response}->content(
 "<HTML><BODY><H1>add task ".join(" ",@job_params)." </H1></BODY></HTM
+L>\n"
     );
    return H_FINAL;
}

sub make_a_worker {
    my ($postback, $message) = @_;     
    POE::Session->create( 
        inline_states => {
            _start    => \&worker_start,
            _stop     => sub {my $session=$_[SESSION]; 
                      print "session ",$session->ID," stop\n";},
            worker_child_stdout => \&worker_child_stdout,
            worker_closed=> \&worker_closed, 
            sig_child           => \&sig_child,
        },
        args          => [ @_ ]
    );
}

sub worker_start
{
 my ($kernel,$session,$heap,@args) = @_[KERNEL, SESSION, HEAP, ARG0 ..
+ $#_];
    $heap->{postback} =$args[0]; 
    $heap->{filename} =$args[1]; 
    
    $heap->{cmdline} = POE::Wheel::Run::Win32->new(
        Program         => executable,
        ProgramArgs     => \@args,
        StdoutEvent     => "worker_child_stdout",
        StderrEvent     => "worker_child_stdout",
        CloseEvent      => "worker_closed",
        StdioFilter  => POE::Filter::Line->new(),  
        StderrFilter => POE::Filter::Line->new()
    );
}

sub worker_child_stdout
{
    my ( $heap, $stdout ) = @_[ HEAP, ARG0 ];

# this code sets $heap->{results}="BUSY"
# when captures this line from the executable's output

}

## NOW THE CLOSEST TO THE QUESTION PART

sub worker_closed
{
    my ( $kernel, $heap ) = @_[ KERNEL, HEAP ];
    my $postback=delete $heap->{postback};
    my @results=($heap->{results});

# skipped code, deleting the intermediate created files

    my $d=delete $heap->{cmdline};
    $postback->(@results); 

    if($results[0] eq "BUSY") {
        print "worker_closed: engine busy, repeat\n";
        my @job_params=($heap->{filename});
        POE::Kernel->post( 'passive', 
                           'enqueue',   
                           'postback',  
                            @job_params); 
    }
}

[download]

In reply to How to enqueue again from the job in POE::Component::JobQueue? by wl2776

Title:
Use:  <p> text here (a paragraph) </p>
and:  <code> code here </code>
to format your post, it's "PerlMonks-approved HTML":


  • Posts are HTML formatted. Put <p> </p> tags around your paragraphs. Put <code> </code> tags around your code and data!
  • Titles consisting of a single word are discouraged, and in most cases are disallowed outright.
  • Read Where should I post X? if you're not absolutely sure you're posting in the right place.
  • Please read these before you post! —
  • Posts may use any of the Perl Monks Approved HTML tags:
    a, abbr, b, big, blockquote, br, caption, center, col, colgroup, dd, del, details, div, dl, dt, em, font, h1, h2, h3, h4, h5, h6, hr, i, ins, li, ol, p, pre, readmore, small, span, spoiler, strike, strong, sub, summary, sup, table, tbody, td, tfoot, th, thead, tr, tt, u, ul, wbr
  • You may need to use entities for some characters, as follows. (Exception: Within code tags, you can put the characters literally.)
            For:     Use:
    & &amp;
    < &lt;
    > &gt;
    [ &#91;
    ] &#93;
  • Link using PerlMonks shortcuts! What shortcuts can I use for linking?
  • See Writeup Formatting Tips and other pages linked from there for more info.