view release on metacpan or  search on metacpan

lib/TAP/Parser/SourceHandler/Worker.pm  view on Meta::CPAN

    $number_of_workers = $source->{config}->{$option_name}->{number_of_workers}
      || 1;
    my $startup         = $source->{config}->{$option_name}->{start_up};
    my $teardown        = $source->{config}->{$option_name}->{tear_down};
    my $error_log       = $source->{config}->{$option_name}->{error_log};
    my $detach          = $source->{config}->{$option_name}->{detach};
    my $sync_type       = $source->{config}->{$option_name}->{sync_type};
    my $source_dir      = $source->{config}->{$option_name}->{source_dir};
    my $destination_dir = $source->{config}->{$option_name}->{destination_dir};
    my %args            = ();
    $args{start_up}        = $startup             if ($startup);
    $args{tear_down}       = $teardown            if ($teardown);
    $args{detach}          = $detach              if ($detach);
    $args{sync_type}       = $sync_type           if ($sync_type);
    $args{source_dir}      = $source_dir          if ($source_dir);
    $args{destination_dir} = $destination_dir     if ($destination_dir);
    $args{error_log}       = $error_log           if ($error_log);
    $args{switches}        = $source->{switches};
    $args{test_args}       = $source->{test_args} if ( $source->{test_args} );

    if ( @workers < $number_of_workers ) {
        my $listener = $class->listener;
        if ( $use_local_public_ip && !$local_public_ip ) {
            require Net::Address::IP::Local;
            $local_public_ip = Net::Address::IP::Local->public;
        }

        my $spec = (
            $local_public_ip
              || (
                $listener->sockhost eq '0.0.0.0'
                ? hostname
                : $listener->sockhost
              )
          )
          . ':'
          . $listener->sockport;
        my $iterator_class = $class->iterator_class;
        eval "use $iterator_class;";
        $args{spec} = $spec;
        my $iterator = $class->iterator_class->new( \%args );
        push @workers, $iterator;
    }
    return $listener->accept();
}

=head3 C<listener>

  my $listener = $class->listener();

Returns worker listener L<IO::Socket::INET>

=cut

sub listener {
    my $class = shift;
    unless ($listener) {
        $listener = IO::Socket::INET->new(
            Listen  => 5,
            Proto   => 'tcp',
            Timeout => 40,
        );
    }
    return $listener;
}

=head3 C<iterator_class>

The class of iterator to use, override if you're sub-classing.  Defaults
to L<TAP::Parser::Iterator::Worker>.

=cut

use constant iterator_class => 'TAP::Parser::Iterator::Worker';

=head3 C<workers>

Returns list of workers.

=cut

sub workers {
    return @workers;
}

=head3 C<get_active_workers>
  
  my @active_workers = $class->get_active_workers;

Returns list of active workers.

=cut

sub get_active_workers {
    my $class   = shift;
    my @workers = $class->workers;
    return unless (@workers);
    my @active;
    for my $worker (@workers) {
        next unless ( $worker && $worker->{sel} );
        my @handles = $worker->{sel}->can_read();
        for my $handle (@handles) {
            if ( $handle == $worker->{err} ) {
                my $error = '';
                if ( $handle->read( $error, 640000 ) ) {
                    chomp($error);
                    print STDERR "Worker with error [$error].\n";

                    #LSF: Close the handle.
                    $handle->close();
                    $worker = undef;
                    last;
                }
            }
        }
        push @active, $worker if ($worker);
    }
    return @active;
}

=head3 C<load_options>