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
|
require "spec_helper"
describe Bunny::Channel do
after :each do
connection.close if connection.open?
end
let(:n) { 200 }
shared_examples "publish confirms" do
context "when publishing with confirms enabled" do
it "increments delivery index" do
ch = connection.create_channel
expect(ch).not_to be_using_publisher_confirmations
ch.confirm_select
expect(ch).to be_using_publisher_confirmations
q = ch.queue("", exclusive: true)
x = ch.default_exchange
n.times do
x.publish("xyzzy", routing_key: q.name)
end
expect(ch.next_publish_seq_no).to eq n + 1
expect(ch.wait_for_confirms).to eq true
sleep 0.25
expect(q.message_count).to eq n
q.purge
ch.close
end
describe "#wait_for_confirms" do
it "should not hang when all the publishes are confirmed" do
ch = connection.create_channel
expect(ch).not_to be_using_publisher_confirmations
ch.confirm_select
expect(ch).to be_using_publisher_confirmations
q = ch.queue("", exclusive: true)
x = ch.default_exchange
n.times do
x.publish("xyzzy", routing_key: q.name)
end
expect(ch.next_publish_seq_no).to eq n + 1
expect(ch.wait_for_confirms).to eq true
sleep 0.25
expect {
Bunny::Timeout.timeout(2) do
expect(ch.wait_for_confirms).to eq true
end
}.not_to raise_error
end
it "raises an error when called on a closed channel" do
ch = connection.create_channel
ch.confirm_select
ch.close
expect {
ch.wait_for_confirms
}.to raise_error(Bunny::ChannelAlreadyClosed)
end
end
context "when some of the messages get nacked" do
it "puts the nacks in the nacked_set" do
ch = connection.create_channel
expect(ch).not_to be_using_publisher_confirmations
ch.confirm_select
expect(ch).to be_using_publisher_confirmations
q = ch.queue("", exclusive: true)
x = ch.default_exchange
n.times do
x.publish("xyzzy", routing_key: q.name)
end
#be sneaky to simulate a nack
nacked_tag = nil
ch.instance_variable_get(:@unconfirmed_set_mutex).synchronize do
expect(ch.unconfirmed_set).to_not be_empty
nacked_tag = ch.unconfirmed_set.reduce(ch.next_publish_seq_no - 1) { |lowest, i| i < lowest ? i : lowest }
ch.handle_ack_or_nack(nacked_tag, false, true)
end
expect(ch.nacked_set).not_to be_empty
expect(ch.nacked_set).to include(nacked_tag)
expect(ch.next_publish_seq_no).to eq n + 1
expect(ch.wait_for_confirms).to eq false
expect(ch.nacked_set).not_to be_empty
expect(ch.nacked_set).to include(nacked_tag)
sleep 0.25
expect(q.message_count).to eq n
q.purge
ch.close
end
end
end
end
context "with a multi-threaded connection" do
let(:connection) do
c = Bunny.new(username: "bunny_gem", password: "bunny_password", vhost: "bunny_testbed", continuation_timeout: 10000)
c.start
c
end
include_examples "publish confirms"
it "returns only when all confirmations for publishes are received" do
ch = connection.create_channel
operations_log = []
operations_log_mutex = Mutex.new
acks_received = Queue.new
log_acks = proc do |tag, _, is_nack|
operations_log_mutex.synchronize do
operation = "#{'n' if is_nack}ack_#{tag}"
operations_log << operation unless operations_log.include?(operation)
end
acks_received << true
end
ch.confirm_select(log_acks)
x = ch.default_exchange
q = ch.temporary_queue
x.publish('msg', routing_key: q.name)
# wait for the confirmation to arrive
acks_received.pop
# artificially simulate a slower ack. the test should work properly even
# without this patch, but it's here just to be sure we catch it.
def (x.channel).handle_ack_or_nack(delivery_tag_before_offset, multiple, nack)
sleep 0.1
super
end
x.publish('msg', routing_key: q.name)
x.publish('msg', routing_key: q.name)
if x.wait_for_confirms
operations_log_mutex.synchronize do
operations_log << 'all_confirmed'
end
end
# wait for all the confirmations to arrive
acks_received.pop
acks_received.pop
expect(operations_log).to eq([
'ack_1',
'ack_2',
'ack_3',
'all_confirmed',
])
end
end
context "with a single-threaded connection" do
let(:connection) do
c = Bunny.new(username: "bunny_gem", password: "bunny_password", vhost: "bunny_testbed", continuation_timeout: 10000, threaded: false)
c.start
c
end
include_examples "publish confirms"
end
end
|
