view release on metacpan or  search on metacpan

lib/Class/Usul/IPC/Cmd.pm  view on Meta::CPAN

# Private functions
my $_child_handler; $_child_handler = sub {
   local $OS_ERROR; # So that waitpid does not step on existing value

   while ((my $child_pid = waitpid -1, WNOHANG) > 0) {
      if (WIFEXITED( $CHILD_ERROR ) and $child_pid > ($CHILD_PID || 0)) {
         $CHILD_PID = $child_pid; $CHILD_ENUM = $CHILD_ERROR;
      }
   }

   $SIG{CHLD} = $_child_handler; # In case of unreliable signals
   return;
};

my $_close_child_io = sub { # In the parent, close the child end of the pipes
   my $pipes = shift;

   close $pipes->[ 0 ]->[ 0 ]; undef $pipes->[ 0 ]->[ 0 ];
   close $pipes->[ 1 ]->[ 1 ]; undef $pipes->[ 1 ]->[ 1 ];
   close $pipes->[ 2 ]->[ 1 ]; undef $pipes->[ 2 ]->[ 1 ];
   close $pipes->[ 3 ]->[ 1 ]; undef $pipes->[ 3 ]->[ 1 ];
   return;
};

my $_drain = sub { # Suck up the output from the child process
   my (%hands, @ready); my $selector = IO::Select->new(); my $i = 0;

   while (defined (my $fh = $_[ $i ])) {
      $selector->add( $fh ); $hands{ fileno $fh } = $_[ $i + 1 ]; $i += 2;
   }

   while (@ready = $selector->can_read) {
      for my $fh (@ready) {
         my $buf; my $bytes_read = sysread $fh, $buf, 64 * 1024;

         if ($bytes_read) { $hands{ fileno $fh }->( "${buf}" ) }
         else { $selector->remove( $fh ); close $fh }
      }
   }

   return;
};

my $_err_handler = sub {
   my ($err, $filtered, $standard) = @_;

   return sub {
      my $buf = shift; defined $buf or return;

      blessed $err     and $err->append( $buf );
      $err eq 'out'    and ${ $filtered } .= $buf;
      $err ne 'null'   and ${ $standard } .= $buf;
      $err eq 'stderr' and emit_to \*STDERR, $buf;
      return;
   }
};

my $_filter_out = sub {
   return join "\n", map    { strip_leader $_ }
                     grep   { not m{ (?: Started | Finished ) }msx }
                     split m{ [\n] }msx, $_[ 0 ];
};

my $_four_nonblocking_pipe_pairs = sub {
   return [ nonblocking_write_pipe_pair, nonblocking_write_pipe_pair,
            nonblocking_write_pipe_pair, nonblocking_write_pipe_pair ];
};

my $_has_shell_meta = sub {
   return (is_arrayref $_[ 0 ] && is_member '|',  $_[ 0 ]) ? TRUE
        : (is_arrayref $_[ 0 ] && is_member '&&', $_[ 0 ]) ? TRUE
        : (                           is_arrayref $_[ 0 ]) ? FALSE
        : (                      $_[ 0 ] =~ m{ [|]    }mx) ? TRUE
        : (                      $_[ 0 ] =~ m{ [&][&] }mx) ? TRUE
                                                           : FALSE;
};

my $_make_socket_pipe = sub {
   socketpair( $_[ 0 ], $_[ 1 ], AF_UNIX, SOCK_STREAM, PF_UNSPEC )
      or throw $EXTENDED_OS_ERROR;
   shutdown  ( $_[ 0 ], 1 );  # No more writing for reader
   shutdown  ( $_[ 1 ], 0 );  # No more reading for writer
   return;
};

my $_out_handler = sub {
   my ($out, $filtered, $standard) = @_;

   return sub {
      my $buf = shift; defined $buf or return;

      blessed $out     and $out->append( $buf );
      $out ne 'null'   and ${ $filtered } .= $buf;
      $out ne 'null'   and ${ $standard } .= $buf;
      $out eq 'stdout' and emit_to \*STDOUT, $buf;
      return;
   }
};

my $_partition_command = sub {
   my $cmd = shift; my $aref = []; my @command = ();

   for my $item (grep { defined && length } @{ $cmd }) {
      if ($item !~ m{ [^\\][\<\>\|\&] }mx) { push @{ $aref }, $item }
      else { push @command, $aref, $item; $aref = [] }
   }

   if ($aref->[ 0 ]) {
      if ($command[ 0 ]) { push @command, $aref }
      else { @command = @{ $aref } }
   }

   return \@command;
};

my $_pipe_handler; $_pipe_handler = sub {
   local $OS_ERROR; # So that wait does not step on existing value

   $CHILD_PID = wait; $CHILD_ENUM = (255 << 8) + 13;
   $SIG{PIPE} = $_pipe_handler;
   return;