Skip to content

Commit 9f61562

Browse files
noxdafoxnoxdafox
committed
test(queue): test multiple submissions
issue: #118
1 parent 01257a3 commit 9f61562

File tree

1 file changed

+44
-59
lines changed

1 file changed

+44
-59
lines changed

‎test/queue_SUITE.erl‎

Lines changed: 44 additions & 59 deletions
Original file line numberDiff line numberDiff line change
@@ -77,41 +77,31 @@ deduplicate_message(Config) ->
7777

7878
%% Deduplication header present
7979
%% String
80-
publish_message(Channel, <<"test">>, "deduplicate-this"),
81-
publish_message(Channel, <<"test">>, "deduplicate-this"),
82-
80+
publish_messages(Channel, <<"test">>, "deduplicate-this", 13),
8381
{#'basic.get_ok'{delivery_tag = Tag1}, _} = amqp_channel:call(Channel, Get),
8482
#'basic.get_empty'{} = amqp_channel:call(Channel, Get),
8583
amqp_channel:cast(Channel, #'basic.ack'{delivery_tag = Tag1}),
8684

87-
%% Integer
88-
publish_message(Channel, <<"test">>, 42),
89-
publish_message(Channel, <<"test">>, 42),
90-
85+
%% %% Integer
86+
publish_messages(Channel, <<"test">>, 42, 3),
9187
{#'basic.get_ok'{delivery_tag = Tag2}, _} = amqp_channel:call(Channel, Get),
9288
#'basic.get_empty'{} = amqp_channel:call(Channel, Get),
9389
amqp_channel:cast(Channel, #'basic.ack'{delivery_tag = Tag2}),
9490

95-
%% Float
96-
publish_message(Channel, <<"test">>, 4.2),
97-
publish_message(Channel, <<"test">>, 4.2),
98-
91+
%% %% Float
92+
publish_messages(Channel, <<"test">>, 4.2, 3),
9993
{#'basic.get_ok'{delivery_tag = Tag3}, _} = amqp_channel:call(Channel, Get),
10094
#'basic.get_empty'{} = amqp_channel:call(Channel, Get),
10195
amqp_channel:cast(Channel, #'basic.ack'{delivery_tag = Tag3}),
10296

103-
%% None/null/nil/void/undefined
104-
publish_message(Channel, <<"test">>, undefined),
105-
publish_message(Channel, <<"test">>, undefined),
106-
97+
%% %% None/null/nil/void/undefined
98+
publish_messages(Channel, <<"test">>, undefined, 3),
10799
{#'basic.get_ok'{delivery_tag = Tag4}, _} = amqp_channel:call(Channel, Get),
108100
#'basic.get_empty'{} = amqp_channel:call(Channel, Get),
109101
amqp_channel:cast(Channel, #'basic.ack'{delivery_tag = Tag4}),
110102

111103
%% Deduplication header absent
112-
publish_message(Channel, <<"test">>),
113-
publish_message(Channel, <<"test">>),
114-
104+
publish_messages(Channel, <<"test">>, 2),
115105
{#'basic.get_ok'{}, _} = amqp_channel:call(Channel, Get),
116106
{#'basic.get_ok'{}, _} = amqp_channel:call(Channel, Get).
117107

@@ -125,17 +115,15 @@ deduplicate_message_ttl(Config) ->
125115
bind_new_exchange(Channel, <<"test">>, <<"test">>),
126116

127117
%% Queue default TTL
128-
publish_message(Channel, <<"test">>, "deduplicate-this"),
118+
publish_messages(Channel, <<"test">>, "deduplicate-this", 1),
129119
timer:sleep(2000),
130-
publish_message(Channel, <<"test">>, "deduplicate-this"),
131-
120+
publish_messages(Channel, <<"test">>, "deduplicate-this", 1),
132121
{#'basic.get_ok'{}, _} = amqp_channel:call(Channel, Get),
133122

134123
%% Message TTL override
135-
publish_message(Channel, <<"test">>, "deduplicate-that", <<"500">>),
124+
publish_messages(Channel, <<"test">>, "deduplicate-that", <<"500">>, 1),
136125
timer:sleep(800),
137-
publish_message(Channel, <<"test">>, "deduplicate-that", <<"500">>),
138-
126+
publish_messages(Channel, <<"test">>, "deduplicate-that", <<"500">>, 1),
139127
{#'basic.get_ok'{}, _} = amqp_channel:call(Channel, Get).
140128

141129
deduplicate_message_confirm(Config) ->
@@ -149,11 +137,10 @@ deduplicate_message_confirm(Config) ->
149137
bind_new_exchange(Channel, <<"test">>, <<"test">>),
150138

151139
%% Publish and wait for confirmation
152-
publish_message(Channel, <<"test">>, "deduplicate-this"),
140+
publish_messages(Channel, <<"test">>, "deduplicate-this", 1),
153141
true = amqp_channel:wait_for_confirms(Channel, 3),
154-
publish_message(Channel, <<"test">>, "deduplicate-this"),
142+
publish_messages(Channel, <<"test">>, "deduplicate-this", 3),
155143
false = amqp_channel:wait_for_confirms(Channel, 3),
156-
157144
{#'basic.get_ok'{}, _} = amqp_channel:call(Channel, Get).
158145

159146
message_acknowledged(Config) ->
@@ -163,14 +150,11 @@ message_acknowledged(Config) ->
163150
#'queue.declare_ok'{} = amqp_channel:call(Channel, make_queue(<<"test">>)),
164151
bind_new_exchange(Channel, <<"test">>, <<"test">>),
165152

166-
publish_message(Channel, <<"test">>, "deduplicate-this"),
167-
153+
%% Acked message does not get deduplicated
154+
publish_messages(Channel, <<"test">>, "deduplicate-this", 1),
168155
{#'basic.get_ok'{delivery_tag = Tag}, _} = amqp_channel:call(Channel, Get),
169-
170156
amqp_channel:cast(Channel, #'basic.ack'{delivery_tag = Tag}),
171-
172-
publish_message(Channel, <<"test">>, "deduplicate-this"),
173-
157+
publish_messages(Channel, <<"test">>, "deduplicate-this", 1),
174158
{#'basic.get_ok'{}, _} = amqp_channel:call(Channel, Get).
175159

176160
queue_overflow(Config) ->
@@ -182,10 +166,10 @@ queue_overflow(Config) ->
182166
make_queue(<<"test">>, Args)),
183167
bind_new_exchange(Channel, <<"test">>, <<"test">>),
184168

185-
publish_message(Channel, <<"test">>, "deduplicate-this"),
186-
publish_message(Channel, <<"test">>, "deduplicate-that"),
187-
publish_message(Channel, <<"test">>, "deduplicate-this"),
188-
169+
%% If queue overflows, same happens for cache
170+
publish_messages(Channel, <<"test">>, "deduplicate-this", 1),
171+
publish_messages(Channel, <<"test">>, "deduplicate-that", 1),
172+
publish_messages(Channel, <<"test">>, "deduplicate-this", 1),
189173
{#'basic.get_ok'{},
190174
#amqp_msg{props = #'P_basic'{headers = [
191175
{<<"x-deduplication-header">>,
@@ -206,15 +190,13 @@ dead_letter(Config) ->
206190
#'queue.declare_ok'{} = amqp_channel:call(Channel, make_queue(<<"test">>, Args)),
207191
bind_new_exchange(Channel, <<"test">>, <<"test">>),
208192

209-
publish_message(Channel, <<"test">>, "deduplicate-this"),
193+
publish_messages(Channel, <<"test">>, "deduplicate-this", 1),
210194
{#'basic.get_ok'{delivery_tag = Tag}, _} = amqp_channel:call(Channel, Get),
211-
212195
amqp_channel:cast(Channel, #'basic.reject'{delivery_tag = Tag, requeue = false}),
213196

214-
publish_message(Channel, <<"test">>, "deduplicate-this"),
197+
publish_messages(Channel, <<"test">>, "deduplicate-this", 1),
215198
{#'basic.get_ok'{}, _} = amqp_channel:call(Channel, Get),
216-
217-
publish_message(Channel, <<"test">>, "deduplicate-this"),
199+
publish_messages(Channel, <<"test">>, "deduplicate-this", 1),
218200
{#'basic.get_ok'{}, _} = amqp_channel:call(Channel, DLGet).
219201

220202
consume_no_ack(Config) ->
@@ -229,14 +211,14 @@ consume_no_ack(Config) ->
229211
#'basic.consume_ok'{} -> ok
230212
end,
231213

232-
publish_message(Channel, <<"test">>, "deduplicate-this"),
214+
publish_messages(Channel, <<"test">>, "deduplicate-this", 1),
233215
receive
234216
{#'basic.deliver'{}, _} -> ok
235217
after 1000 ->
236218
error(message_not_received)
237219
end,
238220

239-
publish_message(Channel, <<"test">>, "deduplicate-this"),
221+
publish_messages(Channel, <<"test">>, "deduplicate-this", 1),
240222
receive
241223
{#'basic.deliver'{}, _} -> ok
242224
after 1000 ->
@@ -254,9 +236,7 @@ queue_policy(Config) ->
254236
%% Wait for policy propagation
255237
timer:sleep(1000),
256238

257-
publish_message(Channel, <<"test">>, "deduplicate-this"),
258-
publish_message(Channel, <<"test">>, "deduplicate-this"),
259-
239+
publish_messages(Channel, <<"test">>, "deduplicate-this", 2),
260240
Get = #'basic.get'{queue = <<"test">>},
261241
{#'basic.get_ok'{delivery_tag = Tag}, _} = amqp_channel:call(Channel, Get),
262242
#'basic.get_empty'{} = amqp_channel:call(Channel, Get),
@@ -266,9 +246,7 @@ queue_policy(Config) ->
266246
#'queue.declare_ok'{} = amqp_channel:call(Channel, #'queue.declare'{queue = <<"test0">>}),
267247
bind_new_exchange(Channel, <<"test0">>, <<"test0">>),
268248

269-
publish_message(Channel, <<"test0">>, "deduplicate-this"),
270-
publish_message(Channel, <<"test0">>, "deduplicate-this"),
271-
249+
publish_messages(Channel, <<"test0">>, "deduplicate-this", 2),
272250
Get0 = #'basic.get'{queue = <<"test0">>},
273251
{#'basic.get_ok'{delivery_tag = Tag0}, _} = amqp_channel:call(Channel, Get0),
274252
#'basic.get_empty'{} = amqp_channel:call(Channel, Get0),
@@ -280,9 +258,7 @@ queue_policy(Config) ->
280258
% Policy is cleared, default arguments are restored
281259
rabbit_ct_broker_helpers:clear_policy(Config, 0, <<"policy-test">>),
282260

283-
publish_message(Channel, <<"test">>, "deduplicate-this"),
284-
publish_message(Channel, <<"test">>, "deduplicate-this"),
285-
261+
publish_messages(Channel, <<"test">>, "deduplicate-this", 2),
286262
Get = #'basic.get'{queue = <<"test">>},
287263
{#'basic.get_ok'{delivery_tag = Tag1}, _} = amqp_channel:call(Channel, Get),
288264
{#'basic.get_ok'{delivery_tag = Tag2}, _} = amqp_channel:call(Channel, Get),
@@ -311,12 +287,17 @@ bind_new_exchange(Ch, Ex, Q) ->
311287
Binding = #'queue.bind'{queue = Q, exchange = Ex, routing_key = <<"#">>},
312288
#'queue.bind_ok'{} = amqp_channel:call(Ch, Binding).
313289

314-
publish_message(Ch, Ex) ->
290+
publish_messages(_Ch, _Ex, 0) ->
291+
ok;
292+
publish_messages(Ch, Ex, N) ->
315293
Publish = #'basic.publish'{exchange = Ex, routing_key = <<"#">>},
316294
Msg = #amqp_msg{payload = <<"payload">>},
317-
amqp_channel:cast(Ch, Publish, Msg).
295+
amqp_channel:cast(Ch, Publish, Msg),
296+
publish_messages(Ch, Ex, N-1).
318297

319-
publish_message(Ch, Ex, D) ->
298+
publish_messages(_Ch, _Ex, _D, 0) ->
299+
ok;
300+
publish_messages(Ch, Ex, D, N) ->
320301
Type = case D of
321302
D when is_integer(D) -> long;
322303
D when is_float(D) -> float;
@@ -326,11 +307,15 @@ publish_message(Ch, Ex, D) ->
326307
Props = #'P_basic'{headers = [{<<"x-deduplication-header">>, Type, D}]},
327308
Publish = #'basic.publish'{exchange = Ex, routing_key = <<"#">>},
328309
Msg = #amqp_msg{props = Props, payload = <<"payload">>},
329-
amqp_channel:cast(Ch, Publish, Msg).
310+
amqp_channel:cast(Ch, Publish, Msg),
311+
publish_messages(Ch, Ex, D, N-1).
330312

331-
publish_message(Ch, Ex, D, E) ->
313+
publish_messages(_Ch, _Ex, _D, _E, 0) ->
314+
ok;
315+
publish_messages(Ch, Ex, D, E, N) ->
332316
Props = #'P_basic'{headers = [{<<"x-deduplication-header">>, longstr, D}],
333317
expiration = E},
334318
Publish = #'basic.publish'{exchange = Ex, routing_key = <<"#">>},
335319
Msg = #amqp_msg{props = Props, payload = <<"payload">>},
336-
amqp_channel:cast(Ch, Publish, Msg).
320+
amqp_channel:cast(Ch, Publish, Msg),
321+
publish_messages(Ch, Ex, D, E, N-1).

0 commit comments

Comments
 (0)