File Coverage

blib/lib/IPC/Open3/Callback.pm

Criterion	Covered	Total	%
statement	137	167	82.0
branch	35	52	67.3
condition	15	31	48.3
subroutine	23	29	79.3
pod	4	5	80.0
total	214	284	75.3

use strict;
use warnings;

package IPC::Open3::Callback;
$IPC::Open3::Callback::VERSION = '1.19';
# ABSTRACT: An extension to IPC::Open3 that will feed out and err to callbacks instead of requiring the caller to handle them.
# PODNAME: IPC::Open3::Callback

use Data::Dumper;
use Exporter qw(import);
use Hash::Util qw(lock_keys);
use IO::Select;
use IO::Socket;
use IPC::Open3;
use Symbol qw(gensym);

use parent qw(Class::Accessor);
__PACKAGE__->follow_best_practice;
__PACKAGE__->mk_accessors(
    qw(out_callback err_callback buffer_output select_timeout buffer_size input_buffer));
__PACKAGE__->mk_ro_accessors(qw(pid last_command last_exit_code));

our @EXPORT_OK = qw(safe_open3);

my $logger;
eval {
    require Log::Log4perl;
    $logger = Log::Log4perl->get_logger('IPC::Open3::Callback');
};
if ($@) {
    require IPC::Open3::Callback::Logger;
    $logger = IPC::Open3::Callback::Logger->get_logger();
}

sub new {
    my ( $class, @args ) = @_;
    return bless( {}, $class )->_init(@args);
}

sub _append_to_buffer {
    my ( $self, $buffer_ref, $data, $flush ) = @_;

    my @lines = split( /\n/, $$buffer_ref . $data, -1 );

    # save the last line in the buffer as it may not yet be a complete line
    $$buffer_ref = $flush ? '' : pop(@lines);

    # return all complete lines
    return @lines;
}

sub _clear_input_buffer {
    my ($self) = shift;
    delete( $self->{input_buffer} );
}

sub DESTROY {
    my ($self) = shift;
    $self->_destroy_child();
}

sub _destroy_child {
    my $self = shift;

    my $pid = $self->get_pid();
    if ($pid) {
        waitpid( $pid, 0 );
        $self->_set_last_exit_code( $? >> 8 );

        $logger->debug(
            sub {
                "Exited '",
                    $self->get_last_command(),
                    "' with code ",
                    $self->get_last_exit_code();
            }
        );
        $self->_set_pid();
    }

    return $self->{last_exit_code};
}

sub _init {
    my ( $self, $args_ref ) = @_;

    $self->{buffer_output}  = undef;
    $self->{buffer_size}    = undef;
    $self->{err_callback}   = undef;
    $self->{input_buffer}   = undef;
    $self->{last_command}   = undef;
    $self->{last_exit_code} = undef;
    $self->{out_callback}   = undef;
    $self->{pid}            = undef;
    $self->{select_timeout} = undef;
    lock_keys( %{$self} );

    if ( defined($args_ref) ) {
        $logger->logdie('parameters must be an hash reference')
            unless ( ( ref($args_ref) ) eq 'HASH' );
        $self->{out_callback}   = $args_ref->{out_callback};
        $self->{err_callback}   = $args_ref->{err_callback};
        $self->{buffer_output}  = $args_ref->{buffer_output};
        $self->{select_timeout} = $args_ref->{select_timeout} || 3;
        $self->{buffer_size}    = $args_ref->{buffer_size} || 1024;
    }
    else {
        $self->{select_timeout} = 3;
        $self->{buffer_size}    = 1024;
    }

    return $self;
}

sub _nix_open3 {
    my ( $in_read, $out_write, $err_write, @command ) = @_;
    my ( $in_write, $out_read, $err_read );

    if ( !$in_read ) {
        $in_read  = gensym();
        $in_write = $in_read;
    }
    if ( !$out_write ) {
        $out_read  = gensym();
        $out_write = $out_read;
    }
    if ( !$err_write ) {
        $err_read  = gensym();
        $err_write = $err_read;
    }

    return ( open3( $in_read, $out_write, $err_write, @command ),
        $in_write, $out_read, $err_read );
}

sub run_command {
    my ( $self, @command ) = @_;

    # if last arg is hashref, its command options not arg...
    my $options = {};
    if ( ref( $command[-1] ) eq 'HASH' ) {
        $options = pop(@command);
    }

    my ($out_callback,   $out_buffer_ref, $err_callback,
        $err_buffer_ref, $buffer_size,    $select_timeout
    );
    $out_callback = $options->{out_callback} || $self->get_out_callback();
    $err_callback = $options->{err_callback} || $self->get_err_callback();
    if ( $options->{buffer_output} || $self->get_buffer_output() ) {
        my $out_temp = '';
        my $err_temp = '';
        $out_buffer_ref = \$out_temp;
        $err_buffer_ref = \$err_temp;
    }
    $buffer_size    = $options->{buffer_size}    || $self->get_buffer_size();
    $select_timeout = $options->{select_timeout} || $self->get_select_timeout();

    $self->_set_last_command( \@command );
    $logger->debug( "Running '", $self->get_last_command(), "'" );
    my ( $pid, $in_fh, $out_fh, $err_fh ) = safe_open3_with(
        $options->{in_handle},
        $options->{out_handle},
        $options->{err_handle}, @command
    );
    $self->_set_pid($pid);

    my $select = IO::Select->new();
    $select->add( $out_fh, $err_fh );
    while ( my @ready = $select->can_read($select_timeout) ) {
        if ( $self->get_input_buffer() ) {
            syswrite( $in_fh, $self->get_input_buffer() );
            $self->_clear_input_buffer();
        }
        foreach my $fh (@ready) {
            my $line;
            my $bytes_read = sysread( $fh, $line, $buffer_size );
            if ( !defined($bytes_read) && !$!{ECONNRESET} ) {
                $logger->error( "sysread failed: ", sub { Dumper(%!) } );
                $logger->logdie( "error in running '", $self->get_last_command(), "': ", $! );
            }
            elsif ( !defined($bytes_read) || $bytes_read == 0 ) {
                $select->remove($fh);
                next;
            }
            else {
                if ( $out_fh && $fh == $out_fh ) {
                    $self->_write_to_callback( $out_callback, $line, $out_buffer_ref, 0 );
                }
                elsif ( $err_fh && $fh == $err_fh ) {
                    $self->_write_to_callback( $err_callback, $line, $err_buffer_ref, 0 );
                }
                else {
                    $logger->logdie('Impossible... somehow got a filehandle I dont know about!');
                }
            }
        }
    }

    # flush buffers
    $self->_write_to_callback( $out_callback, '', $out_buffer_ref, 1 );
    $self->_write_to_callback( $err_callback, '', $err_buffer_ref, 1 );

    return $self->_destroy_child();
}

sub safe_open3 {
    return safe_open3_with( undef, undef, undef, @_ );
}

sub safe_open3_with {
    my ( $in_handle, $out_handle, $err_handle, @command ) = @_;

    my @args = (
        $in_handle  ? '<&' . fileno($in_handle)  : undef,
        $out_handle ? '>&' . fileno($out_handle) : undef,
        $err_handle ? '>&' . fileno($err_handle) : undef, @command
    );
    return ( $^O =~ /MSWin32/ ) ? _win_open3(@args) : _nix_open3(@args);
}

sub send_input {
    my ($self) = @_;
    $self->set_input_buffer(shift);
}

sub _set_last_command {
    my ( $self, $command_ref ) = @_;

    $logger->logdie('the command parameter must be an array reference')
        unless ( ( ref($command_ref) ) eq 'ARRAY' );

    $self->{last_command} = join( ' ', @{$command_ref} );
}

sub _set_last_exit_code {
    my ( $self, $code ) = @_;
    $self->{last_exit_code} = $code;
}

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

    if ( !defined($pid) ) {
        delete( $self->{pid} );
    }
    elsif ( $pid !~ /^\d+$/ ) {
        $logger->logdie('the parameter must be an integer');
    }
    else {
        $self->{pid} = $pid;
    }
}

sub _win_open3 {
    my ( $in_read, $out_write, $err_write, @command ) = @_;

    my ($in_pipe_read,   $in_pipe_write, $out_pipe_read,
        $out_pipe_write, $err_pipe_read, $err_pipe_write
    );
    if ( !$in_read ) {
        ( $in_pipe_read, $in_pipe_write ) = _win_pipe();
        $in_read = '>&' . fileno($in_pipe_read);
    }
    if ( !$out_write ) {
        ( $out_pipe_read, $out_pipe_write ) = _win_pipe();
        $out_write = '<&' . fileno($out_pipe_write);
    }
    if ( !$err_write ) {
        ( $err_pipe_read, $err_pipe_write ) = _win_pipe();
        $err_write = '<&' . fileno($err_pipe_write);
    }

    my $pid = open3( $in_read, $out_write, $err_write, @command );

    return ( $pid, $in_pipe_write, $out_pipe_read, $err_pipe_read );
}

sub _win_pipe {
    my ( $read, $write ) = IO::Socket->socketpair( AF_UNIX, SOCK_STREAM, PF_UNSPEC );
    $read->shutdown(SHUT_WR);     # No more writing for reader
    $write->shutdown(SHUT_RD);    # No more reading for writer

    return ( $read, $write );
}

sub _write_to_callback {
    my ( $self, $callback, $data, $buffer_ref, $flush ) = @_;

    return if ( !defined($callback) );

    my $pid = $self->get_pid();
    if ( !defined($buffer_ref) ) {
        &{$callback}( $data, $pid );
        return;
    }

    foreach my $line ( $self->_append_to_buffer( $buffer_ref, $data, $flush ) ) {
        &{$callback}( $line, $pid );
    }
}

1;

__END__