File Coverage

blib/lib/AnyEvent/Proc.pm

Criterion	Covered	Total	%
statement	317	502	63.1
branch	62	126	49.2
condition	13	50	26.0
subroutine	78	148	52.7
pod	36	36	100.0
total	506	862	58.7

use strict;
use warnings;

package AnyEvent::Proc;

# ABSTRACT: Run external commands

use AnyEvent;
use AnyEvent::Handle;
use AnyEvent::Util ();
use Try::Tiny;
use Class::Load;
use Exporter qw(import);
use Carp;
use POSIX;

our $VERSION = '0.105';    # VERSION

our @EXPORT_OK = qw(run run_cb reader writer);

sub _rpipe {
    my ( $R, $W ) = AnyEvent::Util::portable_pipe;
    (
        $R,
        AnyEvent::Handle->new(
            fh       => $W,
            on_error => sub {
                my ( $handle, $fatal, $message ) = @_;
                AE::log warn => "error writing to handle: $message";
                $handle->destroy;
            },
        ),
    );
}

sub _wpipe {
    my ( $R, $W ) = AnyEvent::Util::portable_pipe;
    (
        AnyEvent::Handle->new(
            fh       => $R,
            on_error => sub {
                my ( $handle, $fatal, $message ) = @_;
                AE::log warn => "error reading from handle: $message"
                  unless $message =~ m{unexpected end-of-file}i;
                $handle->destroy;
            },
        ),
        $W,
    );
}

sub _on_read_helper {
    my ( $aeh, $sub ) = @_;
    $aeh->on_read(
        sub {
            my $x = $_[0]->rbuf;
            $_[0]->rbuf = '';
            $sub->($x);
        }
    );
}

sub _read_on_scalar {
    my ( $var, $sub ) = @_;
    my $old = $$var || '';
    tie $$var, __PACKAGE__ . '::TiedScalar', $sub;
    $$var = $old;
    $var;
}

sub _reaper {
    my ( $self, $waiters ) = @_;
    my $sub = sub {

        # my $message = shift; # currently unused
        foreach my $waiter (@$waiters) {
            AE::log debug => "reap $waiter";
            if ( ref $waiter eq 'CODE' ) {
                $waiter->(undef);
            }
            elsif ( ref $waiter eq 'AnyEvent::CondVar' ) {
                $waiter->send(undef);
            }
            elsif ( ref $waiter eq 'Coro::Channel' ) {
                $waiter->shutdown;
            }
            else {
                AE::log note => "cannot reap $waiter";
            }
        }
    };
    push @{ $self->{reapers} } => $sub;
    $sub;
}

sub _push_waiter {
    my ( $self, $what, $var ) = @_;
    push @{ $self->{waiters}->{$what} } => $var;
}

sub _run_cmd {
    my ( $cmd, $redir, $pidref ) = @_;

    my $cv = AE::cv;

    my %redir = %$redir;

    my $pid = fork;
    AE::log error => "cannot fork: $!" unless defined $pid;

    unless ($pid) {

        # move any existing fd's out of the way
        # this also ensures that dup2 is never called with fd1==fd2
        # so the cloexec flag is always cleared
        my ( @oldfh, @close );
        for my $fh ( values %redir ) {
            push @oldfh, $fh;    # make sure we keep it open
            $fh = fileno $fh;    # we only want the fd

            # dup if we are in the way
            # if we "leak" fds here, they will be dup2'ed over later
            defined( $fh = POSIX::dup($fh) )
              or POSIX::_exit(124)
              while exists $redir{$fh};
        }

        # execute redirects
        while ( my ( $k, $v ) = each %redir ) {
            defined POSIX::dup2( $v, $k )
              or POSIX::_exit(123);
        }

        AnyEvent::Util::close_all_fds_except( keys %redir );

        my $bin = $cmd->[0];

        no warnings;    ## no critic

        exec {$bin} @$cmd;

        POSIX::_exit(126);
    }

    $$pidref = $pid;

    my $w;
    $w = AE::child $pid => sub {
        my $status   = $_[1] >> 8;
        my $signal   = $_[1] & 127;
        my $coredump = $_[1] & 128;
        AE::log info  => "child exited with status $status" if $status;
        AE::log debug => "child exited with signal $signal" if $signal;
        AE::log note  => "child exited with coredump"       if $coredump;
        undef $w;
        map { close $_ } values %redir;
        $cv->send($status);
    };

    $cv;
}

sub new {
    my ( $class, %options ) = @_;

    $options{args} ||= [];

    my ( $rIN,  $wIN )  = _rpipe;
    my ( $rOUT, $wOUT ) = _wpipe;
    my ( $rERR, $wERR ) = _wpipe;

    my @xhs = @{ delete( $options{extras} ) || [] };

    my @args = map { "$_" } @{ delete $options{args} };

    my $pid;

    my %redir = (
        0 => $rIN,
        1 => $wOUT,
        2 => $wERR,
        map { ( "$_" => $_->B ) } @xhs
    );

    my $cv = _run_cmd( [ delete $options{bin} => @args ], \%redir, \$pid );
    my $waiter = AE::cv;

    my $self = bless {
        handles => {
            in  => $wIN,
            out => $rOUT,
            err => $rERR,
            map { ( "$_" => $_->A ) } @xhs,
        },
        pid       => $pid,
        listeners => {
            exit       => delete $options{on_exit},
            ttl_exceed => delete $options{on_ttl_exceed},
        },
        eol     => "\n",
        cv      => $cv,
        alive   => 1,
        waiter  => $waiter,
        waiters => {
            in  => [],
            out => [],
            err => [],
            map { ( "$_" => [] ) } @xhs
        },
        reapers => [],
      } => ref $class
      || $class;

    map { $_->{proc} = $self } @xhs;

    {
        my $eol = quotemeta $self->_eol;
        $self->{reol} = delete $options{reol} || qr{$eol};
    }

    if ( $options{ttl} ) {
        $self->{timer} = AnyEvent->timer(
            after => delete $options{ttl},
            cb    => sub {
                return unless $self->alive;
                $self->kill;
                $self->_emit('ttl_exceed');
            }
        );
    }

    my $kill = sub { $self->end };

    if ( $options{timeout} ) {
        $wIN->timeout( $options{timeout} );
        $rOUT->timeout( $options{timeout} );
        $rERR->timeout( $options{timeout} );
        delete $options{timeout};

        $self->_on( timeout => ( delete( $options{on_timeout} ) || $kill ) );
        my $cb = sub { $self->_emit('timeout') };
        $wIN->on_timeout($cb);
        $rOUT->on_timeout($cb);
        $rERR->on_timeout($cb);
    }

    if ( $options{wtimeout} ) {
        $wIN->wtimeout( delete $options{wtimeout} );

        $self->_on( wtimeout => ( delete( $options{on_wtimeout} ) || $kill ) );
        my $cb = sub { $self->_emit('wtimeout') };
        $wIN->on_wtimeout($cb);
    }

    if ( $options{rtimeout} ) {
        $rOUT->rtimeout( delete $options{rtimeout} );

        $self->_on( rtimeout => ( delete( $options{on_rtimeout} ) || $kill ) );
        my $cb = sub { $self->_emit('rtimeout') };
        $rOUT->on_rtimeout($cb);
    }

    if ( $options{etimeout} ) {
        $rERR->rtimeout( delete $options{etimeout} );

        $self->_on( etimeout => ( delete( $options{on_etimeout} ) || $kill ) );
        my $cb = sub { $self->_emit('etimeout') };
        $rERR->on_rtimeout($cb);
    }

    if ( $options{errstr} ) {
        my $sref = delete $options{errstr};
        $$sref = '';
        $self->pipe( err => $sref );
    }

    if ( $options{outstr} ) {
        my $sref = delete $options{outstr};
        $$sref = '';
        $self->pipe( out => $sref );
    }

    $waiter->begin;
    $cv->cb(
        sub {
            $self->{status} = shift->recv;
            $self->{alive}  = 0;
            undef $self->{timer};
            $waiter->end;
            $self->_emit( exit => $self->{status} );
        }
    );

    if ( keys %options ) {
        AE::log note => "unknown left-over option(s): " . join ', ' =>
          keys %options;
    }

    $self;
}

sub reader {
    my ( $r, $w ) = _wpipe;
    bless {
        r      => $r,
        w      => $w,
        fileno => fileno( $r->fh )
      } => __PACKAGE__
      . '::R';
}

sub writer {
    my ( $r, $w ) = _rpipe;
    bless {
        r      => $r,
        w      => $w,
        fileno => fileno( $w->fh )
      } => __PACKAGE__
      . '::W';
}

sub run {
    my $cv = AE::cv;
    run_cb(
        @_,
        sub {
            $cv->send( \@_ );
        }
    )->recv;
    my ( $out, $err, $status ) = @{ $cv->recv };
    $? = $status << 8;
    if (wantarray) {
        return ( $out, $err );
    }
    else {
        carp $err if $err;
        return $out;
    }
}

sub run_cb {
    my $bin  = shift;
    my $cb   = pop;
    my @args = @_;
    my ( $out, $err ) = ( '', '' );
    my $proc = __PACKAGE__->new(
        bin    => $bin,
        args   => \@args,
        outstr => \$out,
        errstr => \$err
    );
    $proc->finish;
    $proc->wait(
        sub {
            my $status = $proc->{status};
            $? = $status << 8;
            $cb->( $out, $err, $status );
        }
    );
}

sub _on {
    my ( $self, $name, $handler ) = @_;
    $self->{listeners}->{$name} = $handler;
}

sub in { shift->_geth('in') }

sub out { shift->_geth('out') }

sub err { shift->_geth('err') }

sub _geth {
    shift->{handles}->{ pop() };
}

sub _eol  { shift->{eol} }
sub _reol { shift->{reol} }

sub _emit {
    my ( $self, $name, @args ) = @_;
    AE::log debug => "trapped $name";
    if ( exists $self->{listeners}->{$name}
        and defined $self->{listeners}->{$name} )
    {
        $self->{listeners}->{$name}->( $self, @args );
    }
}

sub pid {
    shift->{pid};
}

sub fire {
    my ( $self, $signal ) = @_;
    $signal = 'TERM' unless defined $signal;
    $signal =~ s{^sig}{}i;
    AE::log debug   => "fire SIG$signal";
    kill uc $signal => $self->pid;
}

sub kill {
    my ($self) = @_;
    $self->fire('kill');
}

sub fire_and_kill {
    my $self   = shift;
    my $cb     = ( ref $_[-1] eq 'CODE' ? pop : undef );
    my $time   = pop;
    my $signal = uc( pop || 'TERM' );
    my $w      = AnyEvent->timer(
        after => $time,
        cb    => sub {
            return unless $self->alive;
            $self->kill;
        }
    );
    $self->fire($signal);
    if ($cb) {
        return $self->wait(
            sub {
                undef $w;
                $cb->(@_);
            }
        );
    }
    else {
        my $exit = $self->wait;
        undef $w;
        return $exit;
    }
}

sub alive {
    my $self = shift;
    return 0 unless $self->{alive};
    $self->fire(0) ? 1 : 0;
}

sub wait {
    my ( $self, $cb ) = @_;

    my $next = sub {
        my $cv = shift;
        $cv->recv;
        waitpid $self->{pid} => 0;
        $cb->( $self->{status} ) if ref $cb eq 'CODE';
        $self->end;
        $self->{status};
    };
    AE::log debug => "waiting for "
      . ( $self->{waiter}->{_ae_counter} ) . " ends";
    if ($cb) {
        $self->{waiter}->cb($next);
        return $self->{waiter};
    }
    else {
        $self->{waiter}->recv;
        return $next->( $self->{waiter} );
    }
}

sub finish {
    my ($self) = @_;
    $self->in->destroy;
    $self;
}

sub end {
    my ($self) = @_;
    map { $_->destroy } values %{ $self->{handles} };
    map { $_->() } @{ $self->{reapers} };
    $self;
}

sub stop_timeout {
    my ($self) = @_;
    $self->in->timeout(0);
    $self->out->timeout(0);
    $self->err->timeout(0);
}

sub stop_wtimeout {
    my ($self) = @_;
    $self->in->wtimeout(0);
}

sub stop_rtimeout {
    my ($self) = @_;
    $self->out->rtimeout(0);
}

sub stop_etimeout {
    my ($self) = @_;
    $self->err->rtimeout(0);
}

sub write {
    my ( $self, $type, @args ) = @_;
    my $ok = 0;
    try {
        $self->_geth('in')->push_write( $type => @args );
        $ok = 1;
    }
    catch {
        AE::log warn => $_;
    };
    $ok;
}

sub writeln {
    my ( $self, @lines ) = @_;
    $self->write( $_ . $self->_eol ) for @lines;
    $self;
}

sub pipe {
    my $self = shift;
    my $peer = pop;
    my $what = ( pop || 'out' );
    if ( ref $what ) {
        $what = "$what";
    }
    else {
        $what = lc $what;
        $what =~ s{^std}{};
    }
    use Scalar::Util qw(blessed);
    my $sub;
    if ( blessed $peer) {
        if ( $peer->isa(__PACKAGE__) ) {
            $sub = sub {
                $peer->write(shift);
              }
        }
        elsif ( $peer->isa('AnyEvent::Handle') ) {
            $sub = sub {
                $peer->push_write(shift);
              }
        }
        elsif ( $peer->isa('Coro::Channel') ) {
            $sub = sub {
                $peer->put(shift);
              }
        }
        elsif ( $peer->can('print') ) {
            $sub = sub {
                $peer->print(shift);
              }
        }
    }
    elsif ( ref $peer eq 'SCALAR' ) {
        $sub = sub {
            $$peer .= shift;
          }
    }
    elsif ( ref $peer eq 'GLOB' ) {
        $sub = sub {
            print $peer shift();
          }
    }
    elsif ( ref $peer eq 'CODE' ) {
        $sub = $peer;
    }
    if ($sub) {
        AE::log debug => "pipe $peer from $what";
        my $aeh = $self->_geth($what);
        $aeh->on_eof(
            sub {
                AE::log debug => "eof: $what";
                shift->destroy;
                $self->{waiter}->end;
            }
        );
        $self->{output}->{$what} = _on_read_helper( $aeh, $sub );
        $self->{waiter}->begin;
    }
    else {
        AE::log fatal => "cannot pipe $peer from $what";
    }
}

sub pull {
    my ( $self, $peer ) = @_;
    $self->{input} = $peer;
    AE::log debug => "pull $peer to stdin";
    use Scalar::Util qw(blessed);
    my $sub;
    if ( blessed $peer) {
        if ( $peer->isa(__PACKAGE__) ) {
            return $peer->pipe($self);
        }
        elsif ( $peer->isa('AnyEvent::Handle') ) {
            $peer->on_eof(
                sub {
                    AE::log debug => "pull($peer)->on_eof";
                    shift->destroy;
                    $self->finish;
                }
            );
            $peer->on_error(
                sub {
                    AE::log error => "pull($peer)->on_error(" . $_[2] . ")";
                    shift->destroy;
                }
            );
            return _on_read_helper(
                $peer,
                sub {
                    AE::log debug => "pull($peer)->on_read";
                    $self->write( shift() );
                }
            );
        }
        elsif ( $peer->isa('IO::Handle') ) {
            return $self->pull( AnyEvent::Handle->new( fh => $peer ) );
        }
        elsif ( $peer->isa('Coro::Channel') ) {
            if ( my $class = load_class('Coro') ) {
                return $class->new(
                    sub {
                        while ( my $x = $peer->get ) {
                            $self->write($x) or last;
                            Coro::cede();
                        }
                        $self->finish;
                    }
                );
            }
        }
    }
    elsif ( ref $peer eq 'SCALAR' ) {
        return _read_on_scalar(
            $peer,
            sub {
                AE::log debug => "pull($peer)->STORE";
                $self->write( shift() );
            }
        );
    }
    elsif ( ref $peer eq 'GLOB' ) {
        return $self->pull( AnyEvent::Handle->new( fh => $peer ) );
    }
    AE::log fatal => "cannot pull $peer to stdin";
}

sub _push_read {
    my ( $self, $what, @args ) = @_;
    my $ok = 0;
    try {
        $self->_geth($what)->push_read(@args);
        $ok = 1;
    }
    catch {
        AE::log note => "cannot push_read from std$what: $_";
    };
    $ok;
}

sub _unshift_read {
    my ( $self, $what, @args ) = @_;
    my $ok = 0;
    try {
        $self->_geth($what)->unshift_read(@args);
        $ok = 1;
    }
    catch {
        AE::log note => "cannot unshift_read from std$what: $_";
    };
    $ok;
}

sub _readline {
    my ( $self, $what, $sub ) = @_;
    $self->_push_read( $what => line => $self->_reol, $sub );
}

sub _readchunk {
    my ( $self, $what, $bytes, $sub ) = @_;
    $self->_push_read( $what => chunk => $bytes => $sub );
}

sub _sub_cb {
    my ($cb) = @_;
    sub { $cb->( $_[1] ) }
}

sub _sub_cv {
    my ($cv) = @_;
    sub { $cv->send( $_[1] ) }
}

sub _sub_ch {
    my ($ch) = @_;
    sub { $ch->put( $_[1] ) }
}

sub _readline_cb {
    my ( $self, $what, $cb ) = @_;
    $self->_push_waiter( $what => $cb );
    $self->_readline( $what => _sub_cb($cb) );
}

sub _readline_cv {
    my ( $self, $what, $cv ) = @_;
    $cv ||= AE::cv;
    $self->_push_waiter( $what => $cv );
    $cv->send unless $self->_readline( $what => _sub_cv($cv) );
    $cv;
}

sub _readline_ch {
    my ( $self, $what, $channel ) = @_;
    unless ($channel) {
        if ( my $class = load_class('Coro::Channel') ) {
            $channel ||= $class->new;
        }
    }
    $self->_push_waiter( $what => $channel );
    $channel->shutdown unless $self->_readline( $what => _sub_ch($channel) );
    $channel;
}

sub _readlines_cb {
    my ( $self, $what, $cb ) = @_;
    $self->_push_waiter( $what => $cb );
    $self->_geth($what)->on_read(
        sub {
            $self->_readline( $what => _sub_cb($cb) );
        }
    );
}

sub _readlines_ch {
    my ( $self, $what, $channel ) = @_;
    unless ($channel) {
        if ( my $class = load_class('Coro::Channel') ) {
            $channel ||= $class->new;
        }
    }
    $self->_push_waiter( $what => $channel );
    $channel->shutdown unless $self->_geth($what)->on_read(
        sub {
            $self->_readline( $what => _sub_ch($channel) );
        }
    );
    $channel;
}

sub _readchunk_cb {
    my ( $self, $what, $bytes, $cb ) = @_;
    $self->_push_waiter( $what => $cb );
    $self->_readchunk( $what, $bytes, _sub_cb($cb) );
}

sub _readchunk_cv {
    my ( $self, $what, $bytes, $cv ) = @_;
    $cv ||= AE::cv;
    $self->_push_waiter( $what => $cv );
    $self->_readchunk( $what, $bytes, _sub_cv($cv) );
    $cv;
}

sub _readchunk_ch {
    my ( $self, $what, $bytes, $channel ) = @_;
    unless ($channel) {
        if ( my $class = load_class('Coro::Channel') ) {
            $channel ||= $class->new;
        }
    }
    $self->_push_waiter( $what => $channel );
    $channel->shutdown unless $self->_readline( $what => _sub_ch($channel) );
    $channel;
}

sub _readchunks_ch {
    my ( $self, $what, $bytes, $channel ) = @_;
    unless ($channel) {
        if ( my $class = load_class('Coro::Channel') ) {
            $channel ||= $class->new;
        }
    }
    $self->_push_waiter( $what => $channel );
    $channel->shutdown unless $self->_geth($what)->on_read(
        sub {
            $self->_readline( $what => _sub_ch($channel) );
        }
    );
    $channel;
}

sub readline_cb {
    my ( $self, $cb ) = @_;
    $self->_readline_cb( out => $cb );
}

sub readline_cv {
    my ( $self, $cv ) = @_;
    $self->_readline_cv( out => $cv );
}

sub readline_ch {
    my ( $self, $ch ) = @_;
    $self->_readline_ch( out => $ch );
}

sub readlines_cb {
    my ( $self, $cb ) = @_;
    $self->_readlines_cb( out => $cb );
}

sub readlines_ch {
    my ( $self, $ch ) = @_;
    $self->_readlines_ch( out => $ch );
}

sub readline {
    shift->readline_cv->recv;
}

sub readline_error_cb {
    my ( $self, $cb ) = @_;
    $self->_readline_cb( err => $cb );
}

sub readline_error_cv {
    my ( $self, $cv ) = @_;
    $self->_readline_cv( err => $cv );
}

sub readline_error_ch {
    my ( $self, $ch ) = @_;
    $self->_readline_ch( err => $ch );
}

sub readlines_error_cb {
    my ( $self, $cb ) = @_;
    $self->_readlines_cb( out => $cb );
}

sub readlines_error_ch {
    my ( $self, $ch ) = @_;
    $self->_readlines_ch( out => $ch );
}

sub readline_error {
    shift->readline_error_cv->recv;
}

# AnyEvent::Impl::Perl has some issues with POSIX::dup.
# This statement solves the problem.
AnyEvent::post_detect {
    AE::child $$ => sub { };
};

1;

package    # hidden
  AnyEvent::Proc::R;

use overload '""' => sub { shift->{fileno} };

sub A { shift->{r} }
sub B { shift->{w} }

sub on_timeout {
    shift->A->on_wtimeout(pop);
}

sub stop_timeout {
    shift->A->stop_wtimeout;
}

sub pipe {
    my ( $self, $peer ) = @_;
    $self->{proc}->pipe( $self => $peer );
}

sub readline_cb {
    my ( $self, $cb ) = @_;
    $self->{proc}->_readline_cb( $self => $cb );
}

sub readline_cv {
    my ( $self, $cv ) = @_;
    $self->{proc}->_readline_cv( $self => $cv );
}

sub readline_ch {
    my ( $self, $ch ) = @_;
    $self->{proc}->_readline_ch( $self => $ch );
}

sub readlines_cb {
    my ( $self, $cb ) = @_;
    $self->{proc}->_readlines_cb( $self => $cb );
}

sub readlines_ch {
    my ( $self, $ch ) = @_;
    $self->{proc}->_readlines_cb( $self => $ch );
}

sub readline {
    shift->readline_cv->recv;
}

1;

package    # hidden
  AnyEvent::Proc::W;

use overload '""' => sub { shift->{fileno} };

use Try::Tiny;

sub A { shift->{w} }
sub B { shift->{r} }

sub finish {
    shift->A->destroy;
}

sub on_timeout {
    shift->A->on_rtimeout(pop);
}

sub stop_timeout {
    shift->A->stop_rtimeout;
}

sub write {
    my ( $self, $type, @args ) = @_;
    my $ok = 0;
    try {
        $self->A->push_write( $type => @args );
        $ok = 1;
    }
    catch {
        AE::log note => $_;
    };
    $ok;
}

sub writeln {
    my ( $self, @lines ) = @_;
    my $eol = $self->{proc}->_eol;
    $self->write( $_ . $eol ) for @lines;
    $self;
}

sub pull { die 'UNIMPLEMENTED' }

1;

package    # hidden
  AnyEvent::Proc::TiedScalar;

use Tie::Scalar;

our @ISA = ('Tie::Scalar');

sub TIESCALAR {
    bless pop, shift;
}

sub FETCH {
    undef;
}

sub STORE {
    shift->(pop);
}

1;

__END__