File Coverage

blib/lib/IPC/Pipeline.pm

Criterion	Covered	Total	%
statement	56	58	96.5
branch	30	40	75.0
condition			n/a
subroutine	8	8	100.0
pod	0	2	0.0
total	94	108	87.0

package IPC::Pipeline;

# Copyright (c) 2012, cPanel, Inc.
# All rights reserved.
# http://cpanel.net/
#
# This is free software; you can redistribute it and/or modify it under the same
# terms as Perl itself.  See the LICENSE file for further details.

use strict;
use warnings;

use POSIX ();

BEGIN {
    use Exporter ();
    use vars qw( $VERSION @ISA @EXPORT );

    $VERSION = '0.9';
    @ISA     = ('Exporter');
    @EXPORT  = ('pipeline');
}

sub exec_filter {
    my ($filter) = @_;

    if ( ref($filter) eq 'CODE' ) {
        exit $filter->();
    }
    elsif ( ref($filter) eq 'ARRAY' ) {
        exec(@$filter) or die("Cannot exec(): $!");
    }

    die('Invalid filter');
}

sub pipeline {
    my @filters = @_[ 3 .. $#_ ];

    die('Not enough arguments') unless @_ >= 4;

    #
    # Validate the filters and die early.
    #
    foreach my $filter (@filters) {
        next if ref($filter) =~ /^CODE|ARRAY$/;

        die('Filter passed is not a CODE reference or ARRAY containing command and arguments');
    }

    #
    # Create the initial pipe for passing data into standard input to the first
    # filter passed.  Share a single pipe for standard error use for each
    # process.
    #
    pipe my ( $child_out, $in )       or die("Cannot create a file handle pair for standard input piping: $!");
    pipe my ( $error_out, $error_in ) or die("Cannot create a file handle pair for standard error piping: $!");

    my @pids;

    foreach my $filter (@filters) {
        pipe my ( $out, $child_in ) or die("Cannot create a file handle pair for standard output piping: $!");

        my $pid = fork();

        if ( !defined($pid) ) {
            die("Cannot fork(): $!");
        }
        elsif ( $pid == 0 ) {
            open( STDIN,  '<&=' . fileno($child_out) ) or die("Cannot dup2() last output fd to current child stdin: $!");
            open( STDOUT, '>&=' . fileno($child_in) )  or die("Cannot dup2() last input fd to current child stdout: $!");
            open( STDERR, '>&=' . fileno($error_in) )  or die("Cannot dup2() error pipe input to current child stderr: $!");

            exec_filter($filter);
        }

        #
        # This last child STDOUT file handle should be duplicated onto the next
        # process' standard input reader, or will be passed as the last child
        # output file descriptor if no other subsequent commands are left
        # to be run.
        #
        $child_out = $out;

        push @pids, $pid;
    }

    #
    # Substitute the first three arguments passed by the user with the file
    # handle on the parent's writing end of the initial pipe created for
    # writing to the first command, the last output file handle for the
    # last command, and the standard error handle.  If typeglobs or numeric
    # file descriptors for existing file handles are passed, an attempt will
    # be made to dup2() them as appropriate.
    #
    # The evals around the assignments are present so that we don't die when the
    # argument passed in is undef.  Perl 5.19 and newer don't like that.
    # [perl #7508, #109726]
    if ( !defined $_[0] ) {
        eval { $_[0] = $in };
    }
    elsif ( ref( $_[0] ) eq 'GLOB' ) {
        open( $_[0], '>&=' . fileno($in) );
    }
    else {
        POSIX::dup2( fileno($in), $_[0] );
    }

    if ( !defined $_[1] ) {
        eval { $_[1] = $child_out };
    }
    elsif ( ref( $_[1] ) eq 'GLOB' ) {
        open( $_[1], '<&=' . fileno($child_out) );
    }
    else {
        POSIX::dup2( fileno($child_out), $_[1] );
    }

    if ( !defined $_[2] ) {
        eval { $_[2] = $error_out };
    }
    elsif ( ref( $_[2] ) eq 'GLOB' ) {
        open( $_[2], '<&=' . fileno($error_out) );
    }
    else {
        POSIX::dup2( fileno($error_out), $_[2] );
    }

    return wantarray ? @pids : \@pids;
}

1;

__END__