Hello,

I'm playing with a toy POE program that implements two sessions connected by a pipe that should simply cat an inupt file. Something like this:

                              +- read side of pipe
                             /
          ----------         ----------
STDIN --> | reader | -pipe-> | writer | --> STDOUT
          ----------         ----------
                    \
                     +- write side of pipe
[download]

I started by using rcaputo's example that he posted here at perlmonks.org/?node_id=949005 and came up with this non-working code:

use strict;
use warnings;

use IO::Pipely qw(pipely);
use POE qw(Wheel::ReadWrite);

my ($read_pipe, $write_pipe) = pipely();

# reader sends output to the write pipe
my $reader = POE::Session->create(
    args => [ $write_pipe ],
    inline_states => {
        _start => sub {
            $_[HEAP]->{wheel} = POE::Wheel::ReadWrite->new(
                InputHandle  => \*STDIN,
                OutputHandle => $_[ARG0],
                InputEvent   => 'got_input',
                ErrorEvent   => 'got_input_error',
            );
        },
        got_input          => \&handle_input,
        got_input_error    => \&handle_input_error,
        process_next_input => \&process_next_input,
    }
);

# writer reads input from the read pipe
my $writer = POE::Session->create(
    args => [$read_pipe],
    inline_states => {
        _start => sub {
            $_[HEAP]->{wheel} = POE::Wheel::ReadWrite->new(
                InputHandle  => $_[ARG0],
                OutputHandle => \*STDOUT,
                InputEvent   => 'got_input',
                ErrorEvent   => 'got_input_error',
            );
        },
        got_input          => \&handle_input,
        got_input_error    => \&handle_input_error,
        process_next_input => \&process_next_input,
    }
);

POE::Kernel->run();
exit;

sub handle_input
{
    my ( $kernel, $heap, $input ) = @_[KERNEL, HEAP, ARG0];
    print $heap->{wheel}->put( $input );
    $heap->{wheel}->flush();
    $heap->{wheel}->pause_input();
    $kernel->yield( 'process_next_input' );
}

sub handle_input_error
{
    my ($kernel, $heap) = @_[KERNEL, HEAP];
    $kernel->yield( 'process_next_input' );
    delete $heap->{wheel};
}

sub process_next_input
{
    my ($kernel, $heap) = @_[KERNEL, HEAP];
    $heap->{wheel}->resume_input() if $heap->{wheel};
}
[download]

On my Mac using Perl 5.20 I get this as the output when I feed the program a three line CSV file:

ombibulous> perl cat.pl < test.csv
00001,2,3,4
0a,b,c,d
0foo,bar,baz
^C
ombibulous>
[download]

The ^C is me yanking the rope on my hung program. The lines should not begin with '0's, I don't know how they are creeping in.

I thought (hoped) that after the $reader's handle_input() subroutine does a put(), pauses_input(), and yield(), then POE would see that there is an input for the $writer session and call its input handler. WRONG! I'm not getting how one session cedes processing to another session. I've seen the examples using posts() but I'm trying pipes instead. Can anyone tell me what I'm doing wrong?

Eventually I'd like to build up a chain of sessions, connected by pipes, where each session does a particular transformation on input data; e.g., reader -> parser -> writer. I've already done this with a single process, parsing and writing within a read loop, and subprocesses, forked process for each step. I'd like to see how it works using POE. Is POE::Wheel::ReadWrite the best package to use if I'm going to use a single process?

Thanks for any help, insights, and/or solutions.