foo.t

use strict;
use warnings FATAL => 'all';
use POE;
use Test::More tests => 2;

{
package WheelWrapper;

use strict;
use warnings FATAL => 'all';
use Carp 'croak';
use POE;
use POE::Filter::Stream;
use POE::Wheel::Run;

require Exporter;
use base 'Exporter';
our @EXPORT      = qw(quickie_run quickie quickie_merged quickie_tee quickie_tee_merged);
our @EXPORT_OK   = @EXPORT;
our %EXPORT_TAGS = (ALL => [@EXPORT]);

our %OBJECTS;

sub new {
    my ($package, %args) = @_;

    my $parent_id = $poe_kernel->get_active_session->ID;
    if (my $self = $OBJECTS{$parent_id}) {
        return $self;
    }

    my $self = bless \%args, $package;
    $self->{parent_id} = $parent_id;
    $OBJECTS{$parent_id} = $self;

    return $self;
}

sub _create_session {
    my ($self) = @_;

    POE::Session->create(
        object_states => [
            $self => [qw(
                _start
                _stop
                _exception
                _create_wheel
                _child_signal
                _child_timeout
                _child_stdin
                _child_stdout
                _child_stderr
                _killall
            )],
        ],
        options => {
            ($self->{debug}   ? (debug   => 1) : ()),
            ($self->{default} ? (default => 1) : ()),
            ($self->{trace}   ? (trace   => 1) : ()),
        },
    );

    return;
}

sub _start {
    my ($kernel, $session, $self) = @_[KERNEL, SESSION, OBJECT];

    my $session_id = $session->ID;
    $self->{session_id} = $session_id;
    $kernel->sig(DIE => '_exception');
    return;
}

sub _stop {
    my $self = $_[OBJECT];
    delete $self->{session_id};
    return;
}

sub run {
    my ($self, %args) = @_;

    croak 'Program parameter not supplied' if !defined $args{Program};

    if ($args{AltFork} && ref $args{Program}) {
        croak 'Program must be a string when AltFork is enabled';
    }

    if ($args{AltFork} && $^O eq 'Win32') {
        croak 'AltFork does not currently work on Win32';
    }

    $self->_create_session() if !defined $self->{session_id};

    my ($exception, $wheel)
        = $poe_kernel->call($self->{session_id}, '_create_wheel', \%args);

    # propagate possible exception from POE::Wheel::Run->new()
    croak $exception if $exception;

    return $wheel->PID;
}

sub _create_wheel {
    my ($kernel, $self, $args) = @_[KERNEL, OBJECT, ARG0];

    my %data;
    for my $arg (qw(AltFork Timeout Input Program Context ProgramArgs
            StdoutEvent StderrEvent ExitEvent ResultEvent Tee Merged)) {
        next if !exists $args->{$arg};
        $data{$arg} = delete $args->{$arg};
    }

    if ($data{AltFork}) {
        my @inc = map { +'-I' => $_ } @INC;
        $data{Program} = [$^X, @inc, '-e', $data{Program}];
    }

    my $wheel;
    eval {
        $wheel = POE::Wheel::Run->new(
            StdinFilter => POE::Filter::Stream->new(),
            StdinEvent  => '_child_stdin',
            StdoutEvent => '_child_stdout',
            StderrEvent => '_child_stderr',
            Program     => $data{Program},
            (defined $data{ProgramArgs}
                ? (ProgramArgs => $data{ProgramArgs})
                : ()
            ),
            ($^O ne 'Win32'
                ? (CloseOnCall => 1)
                : ()
            ),
            %$args,
        );
    };

    if ($@) {
        chomp $@;
        return $@;
    }

    $data{obj} = $wheel;
    $data{extra_args} = $args;
    $self->{wheels}{$wheel->ID} = \%data;

    if (defined $data{Input}) {
        $wheel->put($data{Input});
    }
    else {
        $wheel->shutdown_stdin();
    }

    if ($data{Timeout}) {
        $data{alrm} =
            $kernel->delay_set('_child_timeout', $data{Timeout}, $wheel->ID);
    }

    $kernel->sig_child($wheel->PID, '_child_signal');

    return (undef, $wheel);
}

sub _exception {
    my ($kernel, $self, $ex) = @_[KERNEL, OBJECT, ARG1];
    chomp $ex->{error_str};
    warn __PACKAGE__.": Event $ex->{event} in session "
        .$ex->{dest_session}->ID." raised exception:\n  $ex->{error_str}\n";
    $kernel->sig_handled();
    return;
}

sub _child_signal {
    my ($kernel, $self, $pid, $status) = @_[KERNEL, OBJECT, ARG1, ARG2];
    my $id = $self->_pid_to_id($pid);

    my $data = $self->{wheels}{$id};

    my $s = $status >> 8;
    if ($s != 0 && !exists $data->{ExitEvent}
            && !exists $data->{ResultEvent}) {
        warn "Child $pid exited with nonzero status $s\n";
    }

    $kernel->alarm_remove($data->{alrm}) if $data->{Timeout};
    if ($data->{lazy}) {
        $self->{lazy}{$id} = {
            merged => $data->{merged},
            stdout => $data->{stdout},
            stderr => $data->{stderr},
            status => $status,
        }
    }
    delete $self->{wheels}{$id};

    if (defined $data->{ExitEvent}) {
        $kernel->call(
            $self->{parent_id},
            $data->{ExitEvent},
            $status,
            $pid,
            (defined $data->{Context}
                ? $data->{Context}
                : ()),
        );
    }

    if (defined $data->{ResultEvent}) {
        $kernel->call(
            $self->{parent_id},
            $data->{ResultEvent},
            $data->{stdout},
            $data->{stderr},
            $data->{merged},
            $status,
            $pid,
            (defined $data->{Context}
                ? $data->{Context}
                : ()),
        );
    }
    return;
}

sub _child_timeout {
    my ($self, $id) = @_[OBJECT, ARG0];
    $self->{wheels}{$id}{obj}->kill();
    return;
}

sub _child_stdin {
    my ($self, $id) = @_[OBJECT, ARG0];
    $self->{wheels}{$id}{obj}->shutdown_stdin();
    return;
}

sub _child_stdout {
    my ($kernel, $self, $output, $id) = @_[KERNEL, OBJECT, ARG0, ARG1];

    my $data = $self->{wheels}{$id};

    if ($data->{lazy} || defined $data->{ResultEvent}) {
        push @{ $data->{merged} }, $output;
        push @{ $data->{stdout} }, $output;

        if ($data->{lazy}{Tee}) {
            print $output, "\n";
        }
    }
    elsif (!exists $data->{StdoutEvent}) {
        print "$output\n";
    }
    elsif (defined (my $event = $data->{StdoutEvent})) {
        my $context = $data->{Context};
        $kernel->call(
            $self->{parent_id},
            $event,
            $output,
            $data->{obj}->PID,
            (defined $context ? $context : ()),
        );
    }

    return;
}

sub _child_stderr {
    my ($kernel, $self, $error, $id) = @_[KERNEL, OBJECT, ARG0, ARG1];

    my $data = $self->{wheels}{$id};

    if ($data->{lazy} || defined $data->{ResultEvent}) {
        push @{ $data->{merged} }, $error;
        push @{ $data->{stderr} }, $error;

        if ($data->{lazy}{Tee}) {
            $data->{lazy}{Merged}
                ? print $error, "\n"
                : warn $error, "\n";
        }
    }
    elsif (!exists $data->{StderrEvent}) {
        warn "$error\n";
    }
    elsif (defined (my $event = $data->{StderrEvent})) {
        my $context = $data->{Context};
        $kernel->call(
            $self->{parent_id},
            $event,
            $error,
            $data->{obj}->PID,
            (defined $context ? $context : ()),
        );
    }

    return;
}

sub _pid_to_id {
    my ($self, $pid) = @_;

    for my $id (keys %{ $self->{wheels} }) {
        return $id if $self->{wheels}{$id}{obj}->PID == $pid;
    }

    return;
}

sub killall {
    my $self = shift;
    $self = POE::Quickie->new() if ref $self ne 'POE::Quickie';
    $poe_kernel->call($self->{session_id}, '_killall', @_);
    return;
}

sub _killall {
    my ($kernel, $self, $signal) = @_[KERNEL, OBJECT, ARG0];

    $kernel->alarm_remove_all();

    for my $id (keys %{ $self->{wheels}}) {
        $self->{wheels}{$id}{obj}->kill($signal);
    }

    return;
}

sub processes {
    my ($self) = @_;
    $self = POE::Quickie->new() if ref $self ne 'POE::Quickie';

    my %wheels;
    for my $id (keys %{ $self->{wheels} }) {
        my $pid = $self->{wheels}{$id}{obj}->PID;
        $wheels{$pid} = $self->{wheels}{$id}{Context};
    }

    return \%wheels;
}

sub _lazy_run {
    my ($self, %args) = @_;

    my %run_args;
    if (@{ $args{RunArgs} } == 1 &&
        (!ref $args{RunArgs}[0] || ref ($args{RunArgs}[0]) =~ /^(?:ARRAY|CODE)$/)) {
        $run_args{Program} = $args{RunArgs}[0];
    }
    else {
        %run_args = @{ $args{RunArgs} };
    }

    my $pid = $self->run(
        %run_args,
        ExitEvent => undef,
        ($args{Tee} ? () : (StderrEvent => undef)),
        ($args{Tee} ? () : (StdoutEvent => undef)),
    );

    my $id = $self->_pid_to_id($pid);
    $self->{wheels}{$id}{lazy} = {
        Tee    => $args{Tee},
        Merged => $args{Merged},
    };

    my $parent_id = $poe_kernel->get_active_session->ID;
    $poe_kernel->refcount_increment($parent_id, __PACKAGE__);
    $poe_kernel->run_one_timeslice() while $self->{wheels}{$id};
    $poe_kernel->refcount_decrement($parent_id, __PACKAGE__);

    my $data = delete $self->{lazy}{$id};
    return $data->{merged}, $data->{status} if $args{Merged};
    return $data->{stdout}, $data->{stderr}, $data->{status};
}

sub quickie_run {
    my %args = @_;
    my $self = POE::Quickie->new();
    return $self->run(%args);
}

sub quickie {
    my @args = @_;
    my $self = POE::Quickie->new();

    return $self->_lazy_run(
        RunArgs => \@args
    );
}

sub quickie_tee {
    my @args = @_;
    my $self = POE::Quickie->new();
    return $self->_lazy_run(
        RunArgs => \@args,
        Tee     => 1,
    );
}

sub quickie_merged {
    my @args = @_;
    my $self = POE::Quickie->new();

    return $self->_lazy_run(
        RunArgs => \@args,
        Merged  => 1,
    );
}

sub quickie_tee_merged {
    my @args = @_;
    my $self = POE::Quickie->new();

    return $self->_lazy_run(
        RunArgs => \@args,
        Tee     => 1,
        Merged  => 1,
    );
}

}

POE::Session->create(
    package_states => [
        (__PACKAGE__) => [qw(
            _start
            stdout
            stderr
        )],
    ],
);

POE::Kernel->run;

sub _start {
    my $heap = $_[HEAP];

    $heap->{wrapper} = WheelWrapper->new(trace => 1);
    $heap->{wrapper}->run(
        Program     => sub { print "foo\n" },
        StdoutEvent => 'stdout',
    );
}

sub stdout {
    my ($heap, $output) = @_[HEAP, ARG0];
    is($output, 'foo', 'Got stdout');

    $heap->{wrapper}->run(
        Program     => sub { warn "bar\n" },
        StderrEvent => 'stderr',
    );
}

sub stderr {
    my ($heap, $error) = @_[HEAP, ARG0];
    is($error, 'bar', 'Got stderr');
}