1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
|
use Test::More;
use Test::TCP;
use AnyEvent::Gearman::Client;
use Time::HiRes qw/sleep/;
eval q{
use Gearman::Worker;
use Gearman::Server;
};
if ($@) {
plan skip_all
=> "Gearman::Worker and Gearman::Server are required to run this test";
}
my $port = empty_port;
my $client;
my $round = 1;
sub run_tests {
$client ||= AnyEvent::Gearman::Client->new(
job_servers => ['127.0.0.1:' . $port],
);
{
my $cv = AnyEvent->condvar;
$client->add_task(
'reverse', 'Hello World!',
on_complete => sub {
$cv->send($_[1]);
},
on_fail => sub {
$cv->send('fail');
},
);
$round == 1 ?
is $cv->recv, reverse('Hello World!'), 'reverse ok'
: is $cv->recv, 'fail', 'fail because server is down';
}
if ($round == 2) {
my $js = $client->job_servers->[0];
ok !$js->connected, 'disconnect ok';
$js->dead_time(0); # trick for reconnect soon
sleep 0.3;
}
{
my $cv = AnyEvent->condvar;
my $task = $client->add_task(
'sum', '3 5',
on_fail => sub {
$cv->send('fail');
},
);
$task->reg_cb( on_complete => sub { $cv->send($_[1]) } );
is($cv->recv, 8, 'sum ok');
}
$round++;
}
my $child = fork;
if (!defined $child) {
die "fork failed: $!";
}
elsif ($child == 0) {
my $server = Gearman::Server->new( port => $port );
$server->start_worker("$^X t/danga_worker.pl -s 127.0.0.1:$port");
Danga::Socket->EventLoop;
}
else {
END { kill 9, $child if $child }
}
sleep 1;
run_tests;
kill 9, $child;
$child = fork;
if (!defined $child) {
die "fork failed: $!";
}
elsif ($child == 0) {
my $server = Gearman::Server->new( port => $port );
$server->start_worker("$^X t/danga_worker.pl -s 127.0.0.1:$port");
Danga::Socket->EventLoop;
}
else {
END { kill 9, $child if $child }
}
sleep 1;
run_tests;
done_testing;
|
