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
|
use strict;
use Test::More;
use t::Redis;
test_redis {
my $sub = shift;
my $port = shift;
my $info = $sub->info->recv;
if($info->{redis_version} lt "1.3.10") {
plan skip_all => "No PUBLISH/SUBSCRIBE support in this Redis version";
}
my $pub = AnyEvent::Redis->new(host => "127.0.0.1", port => $port);
# $pub is for publishing
# $sub is for subscribing
my $x = 0;
my $expected_x = 0;
my $count = 0;
my $expected_count = 10;
$sub->all_cv->begin;
$sub->subscribe("test.1", sub {
my($message, $chan) = @_;
$x += $message;
if(++$count == $expected_count) {
$sub->unsubscribe("test.1");
is $x, $expected_x, "Messages received, values as expected";
}
});
# Pattern subscription
my $y = 0;
my $expected_y = 0;
my $count2 = 0;
$sub->psubscribe("testp.*", sub {
my($message, $chan) = @_;
$y += $message;
if(++$count2 == $expected_count) {
$sub->punsubscribe("testp.*");
is $y, $expected_y, "Messages received, values as expected";
}
});
for(1 .. $expected_count) {
my $cv = $pub->publish("test.1" => $_);
$expected_x += $_;
# Need to be sure a client has subscribed
$expected_x = 0, redo unless $cv->recv;
}
for(1 .. $expected_count) {
my $cv = $pub->publish("testp.$_" => $_);
$expected_y += $_;
# Need to be sure a client has subscribed
$expected_y = 0, redo unless $cv->recv;
}
$sub->all_cv->end;
done_testing;
};
|
