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
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
|
# encoding: utf-8
require 'spec_helper'
describe "Headers exchange" do
#
# Environment
#
include EventedSpec::AMQPSpec
default_timeout 5
amqp_before do
@connection = AMQP.connect
@channel = AMQP::Channel.new(@connection)
@channel.on_error do |ch, channel_close|
fail "A channel-level exception: #{channel_close.inspect}"
end
end
after(:all) do
AMQP.cleanup_state
done
end
#
# Examples
#
# it would be following good practices to split this into
# 2 separate examples but I think this particular example
# is complete because it demonstrates routing in cases when
# different queues are bound with x-match = any AND x-match = all. MK.
it "can route messages based on any or all of N headers" do
exchange = @channel.headers("amq.match", :durable => true)
linux_and_ia64_messages = []
@channel.queue("", :auto_delete => true).bind(exchange, :arguments => { 'x-match' => 'all', :arch => "IA64", :os => 'linux' }).subscribe do |metadata, payload|
linux_and_ia64_messages << [metadata, payload]
end
linux_and_x86_messages = []
@channel.queue("", :auto_delete => true).bind(exchange, :arguments => { 'x-match' => 'all', :arch => "x86", :os => 'linux' }).subscribe do |metadata, payload|
linux_and_x86_messages << [metadata, payload]
end
any_linux_messages = []
@channel.queue("", :auto_delete => true).bind(exchange, :arguments => { "x-match" => "any", :os => 'linux' }).subscribe do |metadata, payload|
any_linux_messages << [metadata, payload]
end
osx_or_octocore_messages = []
@channel.queue("", :auto_delete => true).bind(exchange, :arguments => { 'x-match' => 'any', :os => 'macosx', :cores => 8 }).subscribe do |metadata, payload|
osx_or_octocore_messages << [metadata, payload]
end
riak_messages = []
@channel.queue("", :auto_delete => true).bind(exchange, :arguments => { "x-match" => "any", :package => { :name => 'riak', :version => '0.14.2' } }).subscribe do |metadata, payload|
riak_messages << [metadata, payload]
end
EventMachine.add_timer(0.5) do
exchange.publish "For linux/IA64", :headers => { :arch => "IA64", :os => 'linux' }
exchange.publish "For linux/x86", :headers => { :arch => "x86", :os => 'linux' }
exchange.publish "For any linux", :headers => { :os => 'linux' }
exchange.publish "For OS X", :headers => { :os => 'macosx' }
exchange.publish "For solaris/IA64", :headers => { :os => 'solaris', :arch => 'IA64' }
exchange.publish "For ocotocore", :headers => { :cores => 8 }
exchange.publish "For nodes with Riak 0.14.2", :headers => { :package => { :name => 'riak', :version => '0.14.2' } }
end
done(4.5) {
linux_and_ia64_messages.size.should == 1
linux_and_x86_messages.size.should == 1
any_linux_messages.size.should == 3
osx_or_octocore_messages.size.should == 2
riak_messages.size.should == 1
}
end
end
describe "Multiple consumers" do
include EventedSpec::AMQPSpec
default_options AMQP_OPTS
default_timeout 5
describe "bound to a queue with the same single header" do
#
# Environment
#
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.headers("amqpgem.tests.integration.headers.exchange1", :auto_delete => true)
@queue.bind(@exchange, :arguments => { :slug => "all", "x-match" => "any" })
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.on_delivery { |metadata, payload| mailbox1 << payload }
consumer2 = AMQP::Consumer.new(@channel, @queue).consume.on_delivery { |metadata, payload| mailbox2 << payload }
EventMachine.add_timer(0.5) do
12.times { @exchange.publish(".", :headers => { :slug => "all" }) }
12.times { @exchange.publish(".", :headers => { :slug => "rspec" }) }
end
done(3.5) {
mailbox1.size.should == 6
mailbox2.size.should == 6
}
end
end
describe "bound to a queue with the same two header & x-match = all" do
#
# Environment
#
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.headers("amqpgem.tests.integration.headers.exchange1", :auto_delete => true)
@queue.bind(@exchange, :arguments => { :slug => "all", :arch => "ia64", 'x-match' => '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.on_delivery { |metadata, payload| mailbox1 << payload }
consumer2 = AMQP::Consumer.new(@channel, @queue).consume.on_delivery { |metadata, payload| mailbox2 << payload }
EventMachine.add_timer(0.5) do
12.times { @exchange.publish(".", :headers => { :slug => "all", :arch => "ia64" }) }
12.times { @exchange.publish(".", :headers => { :slug => "rspec", :arch => "ia64" }) }
end
done(3.5) {
mailbox1.size.should == 6
mailbox2.size.should == 6
}
end
end
describe "bound to 2 queues with the same two header & x-match = all" do
#
# Environment
#
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.headers("amqpgem.tests.integration.headers.exchange1", :auto_delete => true)
args = { :slug => "all", :arch => "ia64", 'x-match' => 'all' }
@queue1.bind(@exchange, :arguments => args)
@queue2.bind(@exchange, :arguments => args)
end
it "get messages distributed to both queues, and in a round-robin manner between consumers on one queue" do
mailbox1 = Array.new
mailbox2 = Array.new
mailbox3 = Array.new
mailbox4 = Array.new
consumer1 = AMQP::Consumer.new(@channel, @queue1).consume.on_delivery { |metadata, payload| mailbox1 << payload }
consumer2 = AMQP::Consumer.new(@channel, @queue1).consume.on_delivery { |metadata, payload| mailbox2 << payload }
consumer3 = AMQP::Consumer.new(@channel, @queue2).consume.on_delivery { |metadata, payload| mailbox3 << payload }
consumer4 = AMQP::Consumer.new(@channel, @queue2).consume.on_delivery { |metadata, payload| mailbox4 << payload }
EventMachine.add_timer(0.5) do
12.times { |i| @exchange.publish("all-#{i}", :headers => { :slug => "all", :arch => "ia64" }) }
16.times { |i| @exchange.publish("rspec-#{i}", :headers => { :slug => "rspec", :arch => "ia64" }) }
end
done(3.5) {
mailbox1.size.should == 6
mailbox1.should == ["all-0", "all-2", "all-4", "all-6", "all-8", "all-10"]
mailbox2.size.should == 6
mailbox2.should == ["all-1", "all-3", "all-5", "all-7", "all-9", "all-11"]
mailbox3.size.should == 6
mailbox4.size.should == 6
}
end
end
describe "bound to a queue with the same two header & x-match = any" do
#
# Environment
#
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.headers("amqpgem.tests.integration.headers.exchange1", :auto_delete => true)
@queue.bind(@exchange, :arguments => { :slug => "all", :arch => "ia64", 'x-match' => 'any' })
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.on_delivery { |metadata, payload| mailbox1 << payload }
consumer2 = AMQP::Consumer.new(@channel, @queue).consume.on_delivery { |metadata, payload| mailbox2 << payload }
EventMachine.add_timer(0.5) do
12.times { @exchange.publish(".", :headers => { :slug => "all", :arch => "ia64" }) }
4.times { @exchange.publish(".", :headers => { :slug => "rspec", :arch => "ia64" }) }
end
done(3.5) {
mailbox1.size.should == 8
mailbox2.size.should == 8
}
end
end
end
|
