AnyEvent-Gearman-WorkerPool

 view release on metacpan or  search on metacpan

lib/AnyEvent/Gearman/WorkerPool.pm  view on Meta::CPAN

    DEBUG __PACKAGE__." start";
    my $self = shift;
    foreach my $key (keys %{$self->slotmap}){
        my $slots = $self->slots($key);
        my $conf = $self->conf($key);
        my $min = $conf->{min};
        foreach my $i ( 0 .. $min-1 ){
            $slots->[$i]->start();
        }
    }
    my $iw = AE::timer 0,5, sub{$self->on_idle;};
    $self->idle_watcher($iw);
}

sub on_idle{
    my $self = shift;
    DEBUG "ON_IDLE";
    foreach my $key (keys %{$self->slotmap}){
        my @slots = @{$self->slots($key)};
        my %conf = %{$self->conf($key)};
        my $idle = 0;

lib/AnyEvent/Gearman/WorkerPool.pm  view on Meta::CPAN

	Log::Log4perl->easy_init($DEBUG);

	use Moose;

	extends 'AnyEvent::Gearman::WorkerPool::Worker';

	sub slowreverse{
        DEBUG 'slowreverse';
        my $self = shift;
        my $job = shift;
        my t = AE::timer 1,0, sub{
            my $res = reverse($job->workload);
            $job->complete( $res );
        };
    }
    sub reverse{
        DEBUG 'reverse';
        my $self = shift;
        my $job = shift;
        my $res = reverse($job->workload);
        DEBUG $res;

lib/AnyEvent/Gearman/WorkerPool/Worker.pm  view on Meta::CPAN

make TestWorker.pm

    package TestWorker;
    use Any::Moose;
    extends 'AnyEvent::Gearman::WorkerPool::Worker';

    sub slowreverse{
        DEBUG 'slowreverse';
        my $self = shift;
        my $job = shift;
        my t = AE::timer 1,0, sub{
            my $res = reverse($job->workload);
            $job->complete( $res );
        };
    }
    sub reverse{
        my $self = shift;
        my $job = shift;
        my $res = reverse($job->workload);
        $job->complete( $res );
    }

t/00-SlotWorker.t  view on Meta::CPAN

use AnyEvent;
use AnyEvent::Gearman;
use TestWorker;
use Log::Log4perl qw(:easy);
# Log::Log4perl->easy_init($DEBUG);

my $port = '9955';
my @js = ("localhost:$port");

my $cv = AE::cv;
my $t = AE::timer 10,0,sub{ $cv->send('timeout')};

use_ok('Gearman::Server');
gstart($port);

my $w = TestWorker->new(job_servers=>\@js, cv => AE::cv, channel=>'test');


my $c = gearman_client @js;

$cv->begin(sub{$cv->send});

t/01-SlotWorkerOverwork.t  view on Meta::CPAN

use AnyEvent;
use AnyEvent::Gearman;
use TestWorker;
use Log::Log4perl qw(:easy);
# Log::Log4perl->easy_init($DEBUG);

my $port = '9955';
my @js = ("localhost:$port");
my $cv = AE::cv;

my $t = AE::timer 10,0,sub{ $cv->send('timeout')};

use_ok('Gearman::Server');
gstart($port);

my $w = TestWorker->new(job_servers=>\@js,cv=>AE::cv,boss_channel=>'', channel=>'test',workleft=>2);


my $c = gearman_client @js;

$cv->begin(sub{

t/02-Slot.t  view on Meta::CPAN

# Log::Log4perl->easy_init($DEBUG);
use Gear;
use AnyEvent;
use AnyEvent::Gearman;
use AnyEvent::Gearman::WorkerPool::Slot;
use Scalar::Util qw(weaken);
my $port = '9955';
my @js = ("localhost:$port");
my $cv = AE::cv;

my $t = AE::timer 10,0,sub{ $cv->send('timeout')};

use_ok('Gearman::Server');
gstart($port);

my $slot = AnyEvent::Gearman::WorkerPool::Slot->new(
    job_servers=>\@js,
    libs=>['t/lib','./lib'],
    workleft=>3,
    worker_package=>'TestWorker',
    worker_channel=>'child',
);

$slot->start();

my $tt = AE::timer 5,0,sub{ 
    $slot->stop();
    is $slot->is_stopped, 1;
    $cv->send;
};

my $res = $cv->recv;
isnt $res,'timeout','ends successfully';
undef($t);
undef($tt);
undef($slot);

t/03-SlotOverwork.t  view on Meta::CPAN

use Gear;
use AnyEvent;
use AnyEvent::Gearman;
use AnyEvent::Gearman::WorkerPool::Slot;
use Scalar::Util qw(weaken);

my $port = '9955';
my @js = ("localhost:$port");
my $cv = AE::cv;

my $t = AE::timer 15,0,sub{ $cv->send('timeout')};

use_ok('Gearman::Server');
gstart($port);

my $slot = AnyEvent::Gearman::WorkerPool::Slot->new(
    job_servers=>\@js,
    libs=>['t/lib','./lib'],
    workleft=>1,
    worker_package=>'TestWorker',
    worker_channel=>'child',

t/04-SlotManager.t  view on Meta::CPAN

use Gear;
use AnyEvent;
use AnyEvent::Gearman;
use AnyEvent::Gearman::WorkerPool;

use Scalar::Util qw(weaken);
my $port = '9955';
my @js = ("localhost:$port");
my $cv = AE::cv;

my $t = AE::timer 10,0,sub{ $cv->send('timeout')};

use_ok('Gearman::Server');
gstart($port);

my $slotman = AnyEvent::Gearman::WorkerPool->new(
    config=>
    {
        global=>{
            job_servers=>\@js,
            libs=>['t/lib','./lib'],
            max=>3,
            },
        slots=>{
            'TestWorker'=>{}
        }
    },
    port=>55595,
);

my $tt = AE::timer 5,0,sub{ 
#    $slot->stop();
#    is $slot->is_stopped, 1;
    $cv->send;
};

my $res = $cv->recv;
isnt $res,'timeout','ends successfully';
undef($tt);
undef($slotman);
gstop();

t/05-SlotManagerWork.t  view on Meta::CPAN

use_ok('Gearman::Server');
gstart($port);

my $cv = AE::cv;

my $sig = AE::signal 'INT'=> sub{ 
DEBUG "TERM!!";
    $cv->send;
};

my $t = AE::timer 10,0,sub{ $cv->send('timeout')};


my $slotman = AnyEvent::Gearman::WorkerPool->new(
    config=>
    {
        global=>{
            job_servers=>\@js,
            libs=>['t/lib','./lib'],
            max=>3,
            },

t/lib/TestWorker.pm  view on Meta::CPAN

use AnyEvent;
use Moose;

extends 'AnyEvent::Gearman::WorkerPool::Worker';

my $t;
sub slowreverse{
    DEBUG 'slowreverse';
    my $self = shift;
    my $job = shift;
    $t = AE::timer 1,0, sub{
        my $res = reverse($job->workload);
        $job->complete( $res );
        undef($t);
    };
}

sub reverse{
    DEBUG 'reverse';
    my $self = shift;
    my $job = shift;