view release on metacpan or  search on metacpan

local/lib/perl5/IO/Async/Process.pm  view on Meta::CPAN

   $self->configure_fd( 2, %{ delete $setup_params{stderr} } ) if $setup_params{stderr};

   $self->configure_fd( 'io', %{ delete $setup_params{stdio} } ) if $setup_params{stdio};

   # All the rest are fd\d+
   foreach ( keys %setup_params ) {
      my ( $fd ) = m/^fd(\d+)$/ or croak "Expected 'fd\\d+'";
      $self->configure_fd( $fd, %{ $setup_params{$_} } );
   }

   $self->SUPER::configure( %params );
}

# These are from the perspective of the parent
use constant FD_VIA_PIPEREAD  => 1;
use constant FD_VIA_PIPEWRITE => 2;
use constant FD_VIA_PIPERDWR  => 3; # Only valid for stdio pseudo-fd
use constant FD_VIA_SOCKETPAIR => 4;

my %via_names = (
   pipe_read  => FD_VIA_PIPEREAD,
   pipe_write => FD_VIA_PIPEWRITE,
   pipe_rdwr  => FD_VIA_PIPERDWR,
   socketpair => FD_VIA_SOCKETPAIR,
);

sub configure_fd
{
   my $self = shift;
   my ( $fd, %args ) = @_;

   $self->is_running and croak "Cannot configure fd $fd in a running Process";

   if( $fd eq "io" ) {
      exists $self->{fd_opts}{$_} and croak "Cannot configure stdio since fd$_ is already defined" for 0 .. 1;
   }
   elsif( $fd == 0 or $fd == 1 ) {
      exists $self->{fd_opts}{io} and croak "Cannot configure fd$fd since stdio is already defined";
   }

   my $opts = $self->{fd_opts}{$fd} ||= {};
   my $via = $opts->{via};

   my ( $wants_read, $wants_write );

   if( my $via_name = delete $args{via} ) {
      defined $via and
         croak "Cannot change the 'via' mode of fd$fd now that it is already configured";

      $via = $via_names{$via_name} or
         croak "Unrecognised 'via' name of '$via_name'";
   }

   if( my $on_read = delete $args{on_read} ) {
      $opts->{handle}{on_read} = $on_read;

      $wants_read++;
   }
   elsif( my $into = delete $args{into} ) {
      $opts->{handle}{on_read} = sub {
         my ( undef, $buffref, $eof ) = @_;
         $$into .= $$buffref if $eof;
         return 0;
      };

      $wants_read++;
   }

   if( defined( my $from = delete $args{from} ) ) {
      $opts->{from} = $from;

      $wants_write++;
   }

   if( defined $via and $via == FD_VIA_SOCKETPAIR ) {
      $self->{fd_opts}{$fd}{$_} = delete $args{$_} for qw( family socktype );
   }

   keys %args and croak "Unexpected extra keys for fd $fd - " . join ", ", keys %args;

   if( !defined $via ) {
      $via = FD_VIA_PIPEREAD  if  $wants_read and !$wants_write;
      $via = FD_VIA_PIPEWRITE if !$wants_read and  $wants_write;
      $via = FD_VIA_PIPERDWR  if  $wants_read and  $wants_write;
   }
   elsif( $via == FD_VIA_PIPEREAD ) {
      $wants_write and $via = FD_VIA_PIPERDWR;
   }
   elsif( $via == FD_VIA_PIPEWRITE ) {
      $wants_read and $via = FD_VIA_PIPERDWR;
   }
   elsif( $via == FD_VIA_PIPERDWR or $via == FD_VIA_SOCKETPAIR ) {
      # Fine
   }
   else {
      die "Need to check fd_via{$fd}\n";
   }

   $via == FD_VIA_PIPERDWR and $fd ne "io" and
      croak "Cannot both read and write simultaneously on fd$fd";

   defined $via and $opts->{via} = $via;
}

sub _prepare_fds
{
   my $self = shift;
   my ( $loop ) = @_;

   my $fd_handle = $self->{fd_handle};
   my $fd_opts   = $self->{fd_opts};

   my $finish_futures = $self->{finish_futures};

   my @setup;

   foreach my $fd ( keys %$fd_opts ) {
      my $opts   = $fd_opts->{$fd};
      my $via    = $opts->{via};

      my $handle = $self->fd( $fd );

local/lib/perl5/IO/Async/Process.pm  view on Meta::CPAN

         push @setup, stdin => [ dup => $childread ], stdout => [ dup => $childwrite ];
         $self->{to_close}{$childread->fileno}  = $childread;
         $self->{to_close}{$childwrite->fileno} = $childwrite;
      }
      elsif( $via == FD_VIA_SOCKETPAIR ) {
         my ( $myfd, $childfd ) = IO::Async::OS->socketpair( $opts->{family}, $opts->{socktype} ) or croak "Unable to socketpair() - $!";

         $handle->configure( handle => $myfd );

         if( $key eq "stdio" ) {
            push @setup, stdin => [ dup => $childfd ], stdout => [ dup => $childfd ];
         }
         else {
            push @setup, $key => [ dup => $childfd ];
         }
         $self->{to_close}{$childfd->fileno} = $childfd;
      }
      else {
         croak "Unsure what to do with fd_via==$via";
      }

      $self->add_child( $handle );

      unless( $write_only ) {
         push @$finish_futures, $handle->new_close_future;
      }
   }

   return @setup;
}

sub _add_to_loop
{
   my $self = shift;
   my ( $loop ) = @_;

   $self->{code} or $self->{command} or
      croak "Require either 'code' or 'command' in $self";

   $self->can_event( "on_finish" ) or
      croak "Expected either an on_finish callback or to be able to ->on_finish";

   my @setup;
   push @setup, @{ $self->{setup} } if $self->{setup};

   push @setup, $self->_prepare_fds( $loop );

   my $finish_futures = delete $self->{finish_futures};

   my ( $exitcode, $dollarbang, $dollarat );
   push @$finish_futures, my $exit_future = $loop->new_future;

   $self->{pid} = $loop->spawn_child(
      code    => $self->{code},
      command => $self->{command},

      setup => \@setup,

      on_exit => $self->_capture_weakself( sub {
         ( my $self, undef, $exitcode, $dollarbang, $dollarat ) = @_;

         $self->debug_printf( "EXIT status=0x%04x", $exitcode ) if $self;
         $exit_future->done unless $exit_future->is_cancelled;
      } ),
   );
   $self->{running} = 1;

   $self->SUPER::_add_to_loop( @_ );

   $_->close for values %{ delete $self->{to_close} };

   my $is_code = defined $self->{code};

   $self->{finish_future} = Future->needs_all( @$finish_futures )
      ->on_done( $self->_capture_weakself( sub {
         my $self = shift or return;

         $self->{exitcode} = $exitcode;
         $self->{dollarbang} = $dollarbang;
         $self->{dollarat}   = $dollarat;

         undef $self->{running};

         if( $is_code ? $dollarat eq "" : $dollarbang == 0 ) {
            $self->invoke_event( on_finish => $exitcode );
         }
         else {
            $self->maybe_invoke_event( on_exception => $dollarat, $dollarbang, $exitcode ) or
               # Don't have a way to report dollarbang/dollarat
               $self->invoke_event( on_finish => $exitcode );
         }

         $self->remove_from_parent;
      } ),
   );
}

sub DESTROY
{
   my $self = shift;
   $self->{finish_future}->cancel if $self->{finish_future};
}

sub notifier_name
{
   my $self = shift;
   if( length( my $name = $self->SUPER::notifier_name ) ) {
      return $name;
   }

   return "nopid" unless my $pid = $self->pid;
   return "[$pid]" unless $self->is_running;
   return "$pid";
}

=head1 METHODS

=cut

=head2 pid

   $pid = $process->pid

Returns the process ID of the process, if it has been started, or C<undef> if
not. Its value is preserved after the process exits, so it may be inspected
during the C<on_finish> or C<on_exception> events.

=cut

sub pid
{
   my $self = shift;
   return $self->{pid};
}

=head2 kill

   $process->kill( $signal )

Sends a signal to the process

=cut

sub kill
{
   my $self = shift;
   my ( $signal ) = @_;

   kill $signal, $self->pid or croak "Cannot kill() - $!";
}

=head2 is_running

   $running = $process->is_running

Returns true if the Process has been started, and has not yet finished.

=cut

sub is_running
{
   my $self = shift;
   return $self->{running};
}

=head2 is_exited

   $exited = $process->is_exited

Returns true if the Process has finished running, and finished due to normal
C<exit(2)>.

=cut

sub is_exited
{
   my $self = shift;
   return defined $self->{exitcode} ? ( $self->{exitcode} & 0x7f ) == 0 : undef;
}

=head2 exitstatus

   $status = $process->exitstatus

If the process exited due to normal C<exit(2)>, returns the value that was
passed to C<exit(2)>. Otherwise, returns C<undef>.

=cut

sub exitstatus
{
   my $self = shift;
   return defined $self->{exitcode} ? ( $self->{exitcode} >> 8 ) : undef;
}

=head2 exception

   $exception = $process->exception

If the process exited due to an exception, returns the exception that was
thrown. Otherwise, returns C<undef>.

=cut

sub exception
{
   my $self = shift;
   return $self->{dollarat};
}

=head2 errno

   $errno = $process->errno

If the process exited due to an exception, returns the numerical value of
C<$!> at the time the exception was thrown. Otherwise, returns C<undef>.

=cut

sub errno
{
   my $self = shift;
   return $self->{dollarbang}+0;
}

=head2 errstr

   $errstr = $process->errstr

If the process exited due to an exception, returns the string value of
C<$!> at the time the exception was thrown. Otherwise, returns C<undef>.

=cut

sub errstr
{
   my $self = shift;
   return $self->{dollarbang}."";
}

=head2 fd

   $stream = $process->fd( $fd )

Returns the L<IO::Async::Stream> or L<IO::Async::Socket> associated with the
given FD number. This must have been set up by a C<configure> argument prior
to adding the C<Process> object to the C<Loop>.

The returned object have its read or write handle set to the other end of a
pipe or socket connected to that FD number in the child process. Typically,
this will be used to call the C<write> method on, to write more data into the
child, or to set an C<on_read> handler to read data out of the child.

The C<on_closed> event for these streams must not be changed, or it will break
the close detection used by the C<Process> object and the C<on_finish> event
will not be invoked.

=cut

sub fd
{
   my $self = shift;
   my ( $fd ) = @_;

   return $self->{fd_handle}{$fd} ||= do {
      my $opts = $self->{fd_opts}{$fd} or
         croak "$self does not have an fd Stream for $fd";

      my $handle_class;
      if( defined $opts->{socktype} && IO::Async::OS->getsocktypebyname( $opts->{socktype} ) != SOCK_STREAM ) {
         require IO::Async::Socket;
         $handle_class = "IO::Async::Socket";
      }
      else {
         require IO::Async::Stream;
         $handle_class = "IO::Async::Stream";
      }

      my $handle = $handle_class->new(
         notifier_name => $fd eq "0"  ? "stdin" :
                          $fd eq "1"  ? "stdout" :
                          $fd eq "2"  ? "stderr" :
                          $fd eq "io" ? "stdio" : "fd$fd",
         %{ $opts->{handle} },
      );

      if( defined $opts->{from} ) {
         $handle->write( $opts->{from},
            on_flush => sub {
               my ( $handle ) = @_;
               $handle->close_write;