view release on metacpan or  search on metacpan

bin/testd  view on Meta::CPAN

#!/usr/bin/env perl

# PODNAME: testd
# ABSTRACT: watch for tests needing running and run them

package Testd;
$Testd::VERSION = '0.001';
use strict;
use warnings;

use POSIX qw{setsid};

use App::Prove::Elasticsearch::Utils;
use MCE::Loop;
use MCE::Signal qw(-setpgrp);
use MCE::Shared;
use List::Util qw{sum};

our $interval        = 5;
our $global_interval = 30;

main() unless caller();

sub main {

    my $conf = App::Prove::Elasticsearch::Utils::process_configuration();

    if (!$conf->{'testd.no_daemon'}) {
        my $pid = daemonify() or die "Could not daemonize";
        print "Spawned as PID $pid\n";
    }

    my $queue = App::Prove::Elasticsearch::Utils::require_queue($conf);
    my $q     = &{\&{$queue . "::new"}}($queue);
    $conf->{'testd.max_workers'} ||= 1;

    my $platformer =
      App::Prove::Elasticsearch::Utils::require_platformer($conf);
    my $versioner = App::Prove::Elasticsearch::Utils::require_versioner($conf);

    my @provisioners;
    if (ref $conf->{'client.provisioners'} eq 'ARRAY') {
        @provisioners = @{$conf->{'client.provisioners'}}
          if ref $conf->{'client.provisioners'} eq 'ARRAY';
    } else {
        push(@provisioners, $conf->{'client.provisioners'})
          if $conf->{'client.provisioners'};
    }
    @provisioners =
      map { App::Prove::Elasticsearch::Utils::require_provisioner($_) }
      @provisioners;

    $0 = "testd - master: waiting for workers to finish";
    print "Testd - starting up...\n";

    while (1) {

        my $jobspec = {};
        $jobspec->{platforms} = &{\&{$platformer . "::get_platforms"}}();
        $jobspec->{version} =
          &{\&{$versioner . "::get_version"}}("$ENV{PWD}/t/bogus.t")
          ; #XXX this will cause trouble with other versioners & planners, I'm sure
        $jobspec->{queue_name} = $q->build_queue_name($jobspec);

        MCE::Shared->start();
        my $worker_state = MCE::Shared->share({module => 'MCE::Shared::Hash'});

        MCE::Loop::init {
            max_workers => $conf->{'testd.max_workers'},
            chunk_size  => 1,
        };

        mce_loop {
            #XXX Net::Rabbitmq is doing something incorrectly, requiring us to re-import every time we fork to avoid 'connection reset by peer'
            my $q_f = &{\&{$queue . "::new"}}($queue);

            #Ensure we have no channel overlap with single-threaded things, like the write_channel, which testd does not use
            $q->{read_channel} = 10 + MCE->wid();
            MCE->say("Worker "
                  . MCE->wid()
                  . " started, checking queue on channel $q->{read_channel}");
            worker($conf, $worker_state, $jobspec, $q_f);