Skip to content

Commit 6de51ee

Browse files
ansdansd
committed
Add queue.declare tests
1 parent d62641a commit 6de51ee

File tree

6 files changed

+246
-66
lines changed

6 files changed

+246
-66
lines changed

‎deps/rabbit/src/rabbit_amqp_management.erl‎

Lines changed: 51 additions & 47 deletions
Original file line numberDiff line numberDiff line change
@@ -63,18 +63,19 @@ handle_http_req(<<"GET">>,
6363
_ConnPid) ->
6464
QNameBin = uri_string:unquote(QNameBinQuoted),
6565
QName = rabbit_misc:r(Vhost, queue, QNameBin),
66-
case rabbit_amqqueue:lookup(QName) of
67-
{ok, Q} ->
68-
{ok, NumMsgs, NumConsumers} = rabbit_amqqueue:stat(Q),
69-
RespPayload = encode_queue(Q, NumMsgs, NumConsumers),
70-
{<<"200">>, RespPayload};
66+
case rabbit_amqqueue:with(
67+
QName,
68+
fun(Q) ->
69+
{ok, NumMsgs, NumConsumers} = rabbit_amqqueue:stat(Q),
70+
RespPayload = encode_queue(Q, NumMsgs, NumConsumers),
71+
{ok, {<<"200">>, RespPayload}}
72+
end) of
73+
{ok, Result} ->
74+
Result;
7175
{error, not_found} ->
72-
case rabbit_amqqueue:not_found_or_absent_dirty(QName) of
73-
not_found ->
74-
throw(<<"404">>, "~ts not found", [rabbit_misc:rs(QName)]);
75-
{absent, Q, Reason} ->
76-
absent(Q, Reason)
77-
end
76+
throw(<<"404">>, "~ts not found", [rabbit_misc:rs(QName)]);
77+
{error, {absent, Q, Reason}} ->
78+
absent(Q, Reason)
7879
end;
7980

8081
handle_http_req(HttpMethod = <<"PUT">>,
@@ -107,44 +108,47 @@ handle_http_req(HttpMethod = <<"PUT">>,
107108
rabbit_core_metrics:queue_declared(QName),
108109

109110
{Q1, NumMsgs, NumConsumers, StatusCode} =
110-
case rabbit_amqqueue:lookup(QName) of
111-
{ok, Q} ->
112-
try rabbit_amqqueue:assert_equivalence(
113-
Q, Durable, AutoDelete, QArgs, Owner) of
114-
ok ->
115-
{ok, Msgs, Consumers} = rabbit_amqqueue:stat(Q),
116-
{Q, Msgs, Consumers, <<"200">>}
117-
catch exit:#amqp_error{name = precondition_failed,
118-
explanation = Expl} ->
119-
throw(<<"409">>, Expl, [])
120-
end;
111+
case rabbit_amqqueue:with(
112+
QName,
113+
fun(Q) ->
114+
try rabbit_amqqueue:assert_equivalence(
115+
Q, Durable, AutoDelete, QArgs, Owner) of
116+
ok ->
117+
{ok, Msgs, Consumers} = rabbit_amqqueue:stat(Q),
118+
{ok, {Q, Msgs, Consumers, <<"200">>}}
119+
catch exit:#amqp_error{name = precondition_failed,
120+
explanation = Expl} ->
121+
throw(<<"409">>, Expl, []);
122+
exit:#amqp_error{explanation = Expl} ->
123+
throw(<<"400">>, Expl, [])
124+
end
125+
end) of
126+
{ok, Result} ->
127+
Result;
121128
{error, not_found} ->
122-
case rabbit_amqqueue:not_found_or_absent_dirty(QName) of
123-
not_found ->
124-
ok = check_vhost_queue_limit(QName),
125-
ok = check_dead_letter_exchange(QName, QArgs, User),
126-
case rabbit_amqqueue:declare(
127-
QName, Durable, AutoDelete, QArgs, Owner, Username) of
128-
{new, Q} ->
129-
rabbit_core_metrics:queue_created(QName),
130-
{Q, 0, 0, <<"201">>};
131-
{absent, Q, Reason} ->
132-
absent(Q, Reason);
133-
{existing, _Q} ->
134-
%% Must have been created in the meantime. Loop around again.
135-
handle_http_req(HttpMethod, PathSegments, Query,
136-
ReqPayload, Vhost, User, ConnPid);
137-
{owner_died, Q} ->
138-
%% Presumably our own days are numbered since the
139-
%% connection has died. Pretend the queue exists though,
140-
%% just so nothing fails.
141-
{Q, 0, 0, <<"201">>};
142-
{protocol_error, _ErrorType, Reason, ReasonArgs} ->
143-
throw(<<"400">>, Reason, ReasonArgs)
144-
end;
129+
ok = check_vhost_queue_limit(QName),
130+
ok = check_dead_letter_exchange(QName, QArgs, User),
131+
case rabbit_amqqueue:declare(
132+
QName, Durable, AutoDelete, QArgs, Owner, Username) of
133+
{new, Q} ->
134+
rabbit_core_metrics:queue_created(QName),
135+
{Q, 0, 0, <<"201">>};
136+
{owner_died, Q} ->
137+
%% Presumably our own days are numbered since the
138+
%% connection has died. Pretend the queue exists though,
139+
%% just so nothing fails.
140+
{Q, 0, 0, <<"201">>};
145141
{absent, Q, Reason} ->
146-
absent(Q, Reason)
147-
end
142+
absent(Q, Reason);
143+
{existing, _Q} ->
144+
%% Must have been created in the meantime. Loop around again.
145+
handle_http_req(HttpMethod, PathSegments, Query,
146+
ReqPayload, Vhost, User, ConnPid);
147+
{protocol_error, _ErrorType, Reason, ReasonArgs} ->
148+
throw(<<"400">>, Reason, ReasonArgs)
149+
end;
150+
{error, {absent, Q, Reason}} ->
151+
absent(Q, Reason)
148152
end,
149153

150154
RespPayload = encode_queue(Q1, NumMsgs, NumConsumers),

‎deps/rabbit/src/rabbit_amqp_session.erl‎

Lines changed: 1 addition & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -48,14 +48,7 @@
4848
%% An even better approach in future would be to dynamically grow (or shrink) the link credit
4949
%% we grant depending on how fast target queue(s) actually confirm messages.
5050
-define(LINK_CREDIT_RCV, 128).
51-
52-
%% Pure HTTP clients of a future HTTP API (v2) would call endpoints as follows:
53-
%% GET /v2/vhosts/:vhost/queues/:queue
54-
%%
55-
%% Here, we use the terminus address /management/v2 so that AMQP 1.0 clients declare the HTTP API version
56-
%% at link attachment time. The vhost is already determined at AMQP connection open time.
57-
%% Therefore, there is no need to send the HTTP API version and the vhost in every HTTP over AMQP request.
58-
-define(MANAGEMENT_NODE_ADDRESS, <<"/management/v2">>).
51+
-define(MANAGEMENT_NODE_ADDRESS, <<"/management">>).
5952

6053
-export([start_link/8,
6154
process_frame/2,

‎deps/rabbit/test/amqp_auth_SUITE.erl‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -55,6 +55,7 @@ groups() ->
5555
declare_queue,
5656
declare_queue_dlx_queue,
5757
declare_queue_dlx_exchange,
58+
declare_queue_vhost_queue_limit,
5859
delete_queue,
5960
purge_queue,
6061
bind_queue_source,
@@ -620,6 +621,24 @@ declare_queue_dlx_exchange(Config) ->
620621
rabbitmq_amqp_client:declare_queue(LinkPair, QName, QProps)),
621622
ok = close_connection_sync(Conn).
622623

624+
declare_queue_vhost_queue_limit(Config) ->
625+
QName = <<"🍿"/utf8>>,
626+
ok = set_permissions(Config, QName, <<>>, <<>>),
627+
Vhost = proplists:get_value(test_vhost, Config),
628+
ok = rabbit_ct_broker_helpers:set_vhost_limit(Config, 0, Vhost, max_queues, 0),
629+
630+
Init = {_, _, LinkPair} = init_pair(Config),
631+
{error, Resp} = rabbitmq_amqp_client:declare_queue(LinkPair, QName, #{}),
632+
?assertMatch(#{subject := <<"403">>}, amqp10_msg:properties(Resp)),
633+
?assertEqual(
634+
#'v1_0.amqp_value'{
635+
content = {utf8, <<"refused to declare queue '", QName/binary, "' in vhost 'test vhost' ",
636+
"because vhost queue limit 0 is reached">>}},
637+
amqp10_msg:body(Resp)),
638+
639+
ok = cleanup_pair(Init),
640+
ok = rabbit_ct_broker_helpers:clear_vhost_limit(Config, 0, Vhost).
641+
623642
delete_queue(Config) ->
624643
{Conn, _, LinkPair} = init_pair(Config),
625644
QName = <<"🍿"/utf8>>,

‎deps/rabbitmq_amqp_client/BUILD.bazel‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -73,6 +73,7 @@ TEST_DEPS = [
7373
rabbitmq_integration_suite(
7474
name = "management_SUITE",
7575
size = "medium",
76+
shard_count = 2,
7677
deps = TEST_DEPS,
7778
)
7879

‎deps/rabbitmq_amqp_client/src/rabbitmq_amqp_client.erl‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,9 @@
1111
-include("rabbitmq_amqp_client.hrl").
1212
-include_lib("amqp10_common/include/amqp10_framing.hrl").
1313

14-
-export[attach_management_link_pair_sync/2,
14+
-export[
15+
%% link pair operations
16+
attach_management_link_pair_sync/2,
1517
detach_management_link_pair_sync/1,
1618

1719
%% queue operations
@@ -30,7 +32,7 @@
3032
].
3133

3234
-define(TIMEOUT, 15_000).
33-
-define(MANAGEMENT_NODE_ADDRESS, <<"/management/v2">>).
35+
-define(MANAGEMENT_NODE_ADDRESS, <<"/management">>).
3436

3537
-type arguments() :: #{binary() => {atom(), term()}}.
3638

0 commit comments

Comments
 (0)