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
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
|
# encoding: utf-8
require 'spec_helper'
describe "Multiple consumers bound to a queue with the same routing key" do
#
# Environment
#
include EventedSpec::AMQPSpec
default_options AMQP_OPTS
default_timeout 5
amqp_before do
@channel = AMQP::Channel.new
@channel.on_error do |ch, close|
raise "Channel-level error!: #{close.inspect}"
end
@queue = @channel.queue("", :auto_delete => true)
@exchange = @channel.direct("amqpgem.tests.integration.direct.exchange", :auto_delete => true)
@queue.bind(@exchange, :routing_key => "builds.all")
end
it "get messages distributed to them in a round-robin manner" do
mailbox1 = Array.new
mailbox2 = Array.new
consumer1 = AMQP::Consumer.new(@channel, @queue).consume
consumer2 = AMQP::Consumer.new(@channel, @queue).consume
consumer1.on_delivery do |metadata, payload|
mailbox1 << payload
end
consumer2.on_delivery do |metadata, payload|
mailbox2 << payload
end
EventMachine.add_timer(0.5) do
12.times { @exchange.publish(".", :routing_key => "builds.all") }
12.times { @exchange.publish(".", :routing_key => "all.builds") }
end
done(4.5) {
mailbox1.size.should == 6
mailbox2.size.should == 6
}
end
end
describe "Multiple queues bound to a direct exchange with the same routing key" do
#
# Environment
#
include EventedSpec::AMQPSpec
default_options AMQP_OPTS
default_timeout 5
amqp_before do
@channel = AMQP::Channel.new
@channel.on_error do |ch, close|
raise "Channel-level error!: #{close.inspect}"
end
@queue1 = @channel.queue("", :auto_delete => true)
@queue2 = @channel.queue("", :auto_delete => true)
@exchange = @channel.direct("amqpgem.tests.integration.direct.exchange", :auto_delete => true)
@queue1.bind(@exchange, :routing_key => "builds.all")
@queue2.bind(@exchange, :routing_key => "builds.all")
end
it "all get a copy of messages with that routing key" do
mailbox1 = Array.new
mailbox2 = Array.new
mailbox3 = Array.new
mailbox4 = Array.new
consumer1 = AMQP::Consumer.new(@channel, @queue1).consume
consumer2 = AMQP::Consumer.new(@channel, @queue1).consume
consumer3 = AMQP::Consumer.new(@channel, @queue2).consume
consumer4 = AMQP::Consumer.new(@channel, @queue2).consume
consumer1.on_delivery do |metadata, payload|
mailbox1 << payload
end
consumer2.on_delivery do |metadata, payload|
mailbox2 << payload
end
consumer3.on_delivery do |metadata, payload|
mailbox3 << payload
end
consumer4.on_delivery do |metadata, payload|
mailbox4 << payload
end
EventMachine.add_timer(0.5) do
13.times { @exchange.publish(".", :routing_key => "builds.all") }
13.times { @exchange.publish(".", :routing_key => "all.builds") }
end
done(3.5) {
mailbox1.size.should == 7
mailbox2.size.should == 6
mailbox3.size.should == 7
mailbox4.size.should == 6
}
end
end
|
