File Coverage

blib/lib/AnyEvent/Open3/Simple.pm

Criterion	Covered	Total	%
statement	113	143	79.0
branch	38	58	65.5
condition	9	13	69.2
subroutine	19	21	90.4
pod	1	2	50.0
total	180	237	75.9

package AnyEvent::Open3::Simple;

use strict;
use warnings;
use 5.006;
use warnings::register;
use IPC::Open3 qw( open3 );
use Scalar::Util qw( reftype );
use Symbol qw( gensym );
use AnyEvent::Open3::Simple::Process;
use Carp qw( croak );
use File::Temp ();
use constant _is_native_win32 => $^O eq 'MSWin32';
use constant _detect => _is_native_win32() ? 'idle' : 'child';

# ABSTRACT: Interface to open3 under AnyEvent
our $VERSION = '0.89'; # VERSION


sub new
{
  my $default_handler = sub { };
  my $class = shift;
  my $args = (reftype($_[0]) || '') eq 'HASH' ? shift : { @_ };
  my %self;
  croak "stdin passed into AnyEvent::Open3::Simple->new no longer supported" if $args->{stdin};
  croak "raw passed into AnyEvent::Open::Simple->new no longer supported" if $args->{raw};
  $self{$_} = $args->{$_} || $default_handler for qw( on_stdout on_stderr on_start on_exit on_signal on_fail on_error on_success );
  $self{impl} = $args->{implementation}
             || $ENV{ANYEVENT_OPEN3_SIMPLE}
             || _detect();
  croak "unknown implementation $self{impl}" unless $self{impl} =~ /^(idle|child|mojo)$/;
  $self{impl} = _detect()
    if $self{impl} eq 'mojo' && do { require Mojo::Reactor; Mojo::Reactor->detect eq 'Mojo::Reactor::EV' };
  bless \%self, $class;
}


sub run
{
  croak "run method requires at least one argument"
    unless @_ >= 2;

  my $proc_user = (ref $_[-1] eq 'CODE' ? pop : sub {});

  my $stdin;
  $stdin = pop if ref $_[-1];

  my($self, $program, @arguments) = @_;

  my($child_stdin, $child_stdout, $child_stderr);
  $child_stderr = gensym;

  local *TEMP;
  if(defined $stdin)
  {
    my $file = File::Temp->new;
    $file->autoflush(1);
    $file->print(
      ref($stdin) eq 'ARRAY'
      ? join("\n", @{ $stdin })
      : $$stdin
    );
    $file->seek(0,0);
    open TEMP, '<&=', $file;  ## no critic
    $child_stdin = '<&TEMP';
  }

  if($self->{impl} =~ /^(child|idle)$/)
  {
    require AnyEvent;
    AnyEvent::detect();
    require AnyEvent::Open3::Simple::Idle if $self->{impl} eq 'idle';
  }
  elsif($self->{impl} eq 'mojo')
  {
    require Mojo::Reactor;
    require Mojo::IOLoop;
    require AnyEvent::Open3::Simple::Mojo;
  }

  my $pid = eval { open3 $child_stdin, $child_stdout, $child_stderr, $program, @arguments };

  if(my $error = $@)
  {
    $self->{on_error}->($error, $program, @arguments);
    return;
  }

  my $proc = AnyEvent::Open3::Simple::Process->new($pid, $child_stdin);
  $proc_user->($proc);

  $self->{on_start}->($proc, $program, @arguments);

  my $watcher_stdout;
  my $watcher_stderr;

  my $stdout_callback = sub {
    my $input = <$child_stdout>;
    return unless defined $input;
    $input =~ s/(\015?\012|\015)$//;
    my $ref = $self->{on_stdout};
    ref($ref) eq 'ARRAY' ? push @$ref, $input : $ref->($proc, $input);
  };

  my $stderr_callback = sub {
    my $input = <$child_stderr>;
    return unless defined $input;
    $input =~ s/(\015?\012|\015)$//;
    my $ref = $self->{on_stderr};
    ref($ref) eq 'ARRAY' ? push @$ref, $input : $ref->($proc, $input);
  };

  if(!_is_native_win32() && $self->{impl} =~ /^(idle|child)$/)
  {
    $watcher_stdout = AnyEvent->io(
      fh   => $child_stdout,
      poll => 'r',
      cb   => $stdout_callback,
    ) unless _is_native_win32();

    $watcher_stderr = AnyEvent->io(
      fh   => $child_stderr,
      poll => 'r',
      cb   => $stderr_callback,
    ) unless _is_native_win32();
  }

  my $watcher_child;

  my $end_cb = sub {
    #my($pid, $status) = @_;
    my $status = $_[1];
    my($exit_value, $signal) = ($status >> 8, $status & 127);

    $proc->close;

    # make sure we consume any stdout and stderr which hasn't
    # been consumed yet.  This seems to make on_out.t work on
    # cygwin
    if($self->{raw})
    {
      local $/;
      $self->{on_stdout}->($proc, scalar <$child_stdout>);
      $self->{on_stderr}->($proc, scalar <$child_stderr>);
    }
    else
    {
      while(!eof $child_stdout)
      {
        my $input = <$child_stdout>;
        last unless defined $input;
        $input =~ s/(\015?\012|\015)$//;
        my $ref = $self->{on_stdout};
        ref($ref) eq 'ARRAY' ? push @$ref, $input : $ref->($proc, $input);
      }

      while(!eof $child_stderr)
      {
        my $input = <$child_stderr>;
        last unless defined $input;
        $input =~ s/(\015?\012|\015)$//;
        my $ref = $self->{on_stderr};
        ref($ref) eq 'ARRAY' ? push @$ref, $input : $ref->($proc, $input);
      }
    }

    $self->{on_exit}->($proc, $exit_value, $signal);
    $self->{on_signal}->($proc, $signal) if $signal > 0;
    $self->{on_fail}->($proc, $exit_value) if $exit_value > 0;
    $self->{on_success}->($proc) if $signal == 0 && $exit_value == 0;
    undef $watcher_stdout;
    undef $watcher_stderr;
    undef $watcher_child;
    undef $proc;
  };

  if($self->{impl} eq 'mojo')
  {
    my($selout, $selerr);

    if(_is_native_win32())
    {
      require IO::Select;
      $selout = IO::Select->new($child_stdout);
      $selerr = IO::Select->new($child_stderr);
    }

    my $reactor = Mojo::IOLoop->singleton->reactor;
    my $id;
    $id = Mojo::IOLoop->recurring(0.25 => sub {
      AnyEvent::Open3::Simple::Mojo::_watcher($pid, sub {
        $end_cb->(@_);
        Mojo::IOLoop->remove($id);
        if(_is_native_win32())
        {
          $stdout_callback->() if $selout->can_read(0);
          $stderr_callback->() if $selerr->can_read(0);
        }
        else
        {
          $reactor->remove($child_stdout);
          $reactor->remove($child_stderr);
        }
      });
    });

  }
  elsif($self->{impl} eq 'idle')
  {
    my($selout, $selerr);

    if(_is_native_win32())
    {
      require IO::Select;
      $selout = IO::Select->new($child_stdout);
      $selerr = IO::Select->new($child_stderr);
    }

    $watcher_child = AnyEvent->idle(cb => sub {
      if(_is_native_win32())
      {
        $stdout_callback->() if $selout->can_read(0);
        $stderr_callback->() if $selerr->can_read(0);
      }
      AnyEvent::Open3::Simple::Idle::_watcher($pid, $end_cb);
    });
  }
  else
  {
    $watcher_child = AnyEvent->child(
      pid => $pid,
      cb  => $end_cb,
    );
  }

  $self;
}

1;

__END__