Skip to content

Commit 0d60fa6

Browse files
lhoguinlhoguin
committed
Mark amqp_address_v1 as denied_by_default
1 parent 2320853 commit 0d60fa6

8 files changed

Lines changed: 101 additions & 61 deletions

File tree

‎deps/amqp10_client/test/system_SUITE.erl‎

Lines changed: 68 additions & 37 deletions
Original file line numberDiff line numberDiff line change
@@ -258,7 +258,7 @@ basic_roundtrip(Config) ->
258258
Hostname = ?config(rmq_hostname, Config),
259259
Port = rabbit_ct_broker_helpers:get_node_config(Config, 0, tcp_port_amqp),
260260
OpenConf = #{address => Hostname, port => Port, sasl => anon},
261-
roundtrip(OpenConf).
261+
roundtrip(Config, OpenConf, ?FUNCTION_NAME).
262262

263263
basic_roundtrip_tls(Config) ->
264264
Hostname = ?config(rmq_hostname, Config),
@@ -274,7 +274,7 @@ basic_roundtrip_tls(Config) ->
274274
notify => self(),
275275
container_id => <<"open_connection_tls_container">>,
276276
sasl => ?config(sasl, Config)},
277-
roundtrip(OpnConf).
277+
roundtrip(Config, OpnConf, ?FUNCTION_NAME).
278278

279279
%% ssl option validation fails if verify_peer is enabled without cacerts.
280280
%% Test that cacertfile option takes effect taken from the application env.
@@ -293,7 +293,7 @@ roundtrip_tls_global_config(Config) ->
293293
notify => self(),
294294
container_id => <<"open_connection_tls_container">>,
295295
sasl => ?config(sasl, Config)},
296-
roundtrip(OpnConf),
296+
roundtrip(Config, OpnConf, ?FUNCTION_NAME),
297297
application:unset_env(amqp10_client, ssl_options).
298298

299299
service_bus_config(Config, ContainerId) ->
@@ -310,10 +310,10 @@ service_bus_config(Config, ContainerId) ->
310310
sasl => {plain, User, Password}}.
311311

312312
basic_roundtrip_service_bus(Config) ->
313-
roundtrip(service_bus_config(Config, <<"basic_roundtrip_service_bus">>)).
313+
roundtrip(Config, service_bus_config(Config, <<"basic_roundtrip_service_bus">>), ?FUNCTION_NAME).
314314

315315
filtered_roundtrip_service_bus(Config) ->
316-
filtered_roundtrip(service_bus_config(Config, <<"filtered_roundtrip_service_bus">>)).
316+
filtered_roundtrip(Config, service_bus_config(Config, <<"filtered_roundtrip_service_bus">>), ?FUNCTION_NAME).
317317

318318
roundtrip_large_messages(Config) ->
319319
Hostname = ?config(rmq_hostname, Config),
@@ -324,19 +324,21 @@ roundtrip_large_messages(Config) ->
324324
DataMb = rand:bytes(1024 * 1024),
325325
Data8Mb = rand:bytes(8 * 1024 * 1024),
326326
Data64Mb = rand:bytes(64 * 1024 * 1024),
327-
ok = roundtrip(OpenConf, DataKb),
328-
ok = roundtrip(OpenConf, DataMb),
329-
ok = roundtrip(OpenConf, Data8Mb),
330-
ok = roundtrip(OpenConf, Data64Mb).
327+
ok = roundtrip(Config, OpenConf, roundtrip_large_messages_kb, DataKb),
328+
ok = roundtrip(Config, OpenConf, roundtrip_large_messages_mb, DataMb),
329+
ok = roundtrip(Config, OpenConf, roundtrip_large_messages_8mb, Data8Mb),
330+
ok = roundtrip(Config, OpenConf, roundtrip_large_messages_64mb, Data64Mb).
331331

332-
roundtrip(OpenConf) ->
333-
roundtrip(OpenConf, <<"banana">>).
332+
roundtrip(Config, OpenConf, FnName) ->
333+
roundtrip(Config, OpenConf, FnName, <<"banana">>).
334334

335-
roundtrip(OpenConf, Body) ->
335+
roundtrip(Config, OpenConf, FnName, Body) ->
336336
{ok, Connection} = amqp10_client:open_connection(OpenConf),
337337
{ok, Session} = amqp10_client:begin_session(Connection),
338+
QName = atom_to_binary(FnName),
339+
Address = declare_queue(Config, QName),
338340
{ok, Sender} = amqp10_client:attach_sender_link(
339-
Session, <<"banana-sender">>, <<"test1">>, settled, unsettled_state),
341+
Session, <<"banana-sender">>, Address, settled, unsettled_state),
340342
await_link(Sender, credited, link_credit_timeout),
341343

342344
Now = os:system_time(millisecond),
@@ -357,7 +359,7 @@ roundtrip(OpenConf, Body) ->
357359

358360
{error, link_not_found} = amqp10_client:detach_link(Sender),
359361
{ok, Receiver} = amqp10_client:attach_receiver_link(
360-
Session, <<"banana-receiver">>, <<"test1">>, settled, unsettled_state),
362+
Session, <<"banana-receiver">>, Address, settled, unsettled_state),
361363
{ok, OutMsg} = amqp10_client:get_msg(Receiver, 4 * 60_000),
362364
ok = amqp10_client:end_session(Session),
363365
ok = amqp10_client:close_connection(Connection),
@@ -370,15 +372,17 @@ roundtrip(OpenConf, Body) ->
370372
?assertEqual([Body], amqp10_msg:body(OutMsg)),
371373
ok.
372374

373-
filtered_roundtrip(OpenConf) ->
374-
filtered_roundtrip(OpenConf, <<"banana">>).
375+
filtered_roundtrip(Config, OpenConf, FnName) ->
376+
filtered_roundtrip(Config, OpenConf, FnName, <<"banana">>).
375377

376-
filtered_roundtrip(OpenConf, Body) ->
378+
filtered_roundtrip(Config, OpenConf, FnName, Body) ->
377379
{ok, Connection} = amqp10_client:open_connection(OpenConf),
378380
{ok, Session} = amqp10_client:begin_session(Connection),
381+
QName = atom_to_binary(FnName),
382+
Address = declare_queue(Config, QName),
379383
{ok, Sender} = amqp10_client:attach_sender_link(Session,
380384
<<"default-sender">>,
381-
<<"test1">>,
385+
Address,
382386
settled,
383387
unsettled_state),
384388
await_link(Sender, credited, link_credit_timeout),
@@ -390,7 +394,7 @@ filtered_roundtrip(OpenConf, Body) ->
390394

391395
{ok, DefaultReceiver} = amqp10_client:attach_receiver_link(Session,
392396
<<"default-receiver">>,
393-
<<"test1">>,
397+
Address,
394398
settled,
395399
unsettled_state),
396400
ok = amqp10_client:send_msg(Sender, Msg1),
@@ -408,7 +412,7 @@ filtered_roundtrip(OpenConf, Body) ->
408412

409413
{ok, FilteredReceiver} = amqp10_client:attach_receiver_link(Session,
410414
<<"filtered-receiver">>,
411-
<<"test1">>,
415+
Address,
412416
settled,
413417
unsettled_state,
414418
#{<<"apache.org:selector-filter:string">> => <<"amqp.annotation.x-opt-enqueuedtimeutc > ", Now2Binary/binary>>}),
@@ -431,8 +435,10 @@ transfer_id_vs_delivery_id(Config) ->
431435

432436
{ok, Connection} = amqp10_client:open_connection(OpenConf),
433437
{ok, Session} = amqp10_client:begin_session(Connection),
438+
QName = <<"test1">>,
439+
Address = declare_queue(Config, QName),
434440
{ok, Sender} = amqp10_client:attach_sender_link(
435-
Session, <<"banana-sender">>, <<"test1">>, settled, unsettled_state),
441+
Session, <<"banana-sender">>, Address, settled, unsettled_state),
436442
await_link(Sender, credited, link_credit_timeout),
437443

438444
P0 = binary:copy(<<0>>, 8_000_000),
@@ -446,7 +452,7 @@ transfer_id_vs_delivery_id(Config) ->
446452
await_link(Sender, {detached, normal}, link_detach_timeout),
447453

448454
{ok, Receiver} = amqp10_client:attach_receiver_link(
449-
Session, <<"banana-receiver">>, <<"test1">>, settled, unsettled_state),
455+
Session, <<"banana-receiver">>, Address, settled, unsettled_state),
450456
{ok, RcvMsg1} = amqp10_client:get_msg(Receiver, 60_000 * 4),
451457
{ok, RcvMsg2} = amqp10_client:get_msg(Receiver, 60_000 * 4),
452458
ok = amqp10_client:end_session(Session),
@@ -513,9 +519,11 @@ early_transfer(Config) ->
513519
Port = rabbit_ct_broker_helpers:get_node_config(Config, 0, tcp_port_amqp),
514520
{ok, Connection} = amqp10_client:open_connection(Hostname, Port),
515521
{ok, Session} = amqp10_client:begin_session(Connection),
522+
QName = <<"test">>,
523+
Address = declare_queue(Config, QName),
516524
{ok, Sender} = amqp10_client:attach_sender_link(Session,
517525
<<"early-transfer">>,
518-
<<"test">>),
526+
Address),
519527

520528
Msg = amqp10_msg:new(<<"my-tag">>, <<"banana">>, true),
521529
% TODO: this is a timing issue - should use mock here really
@@ -526,7 +534,7 @@ early_transfer(Config) ->
526534
% attach then immediately detach
527535
LinkName = <<"early-transfer2">>,
528536
{ok, Sender2} = amqp10_client:attach_sender_link(Session, LinkName,
529-
<<"test">>),
537+
Address),
530538
{error, half_attached} = amqp10_client:detach_link(Sender2),
531539
await_link(Sender2, credited, credited_timeout),
532540
ok = amqp10_client:end_session(Session),
@@ -542,15 +550,17 @@ split_transfer(Config) ->
542550
sasl => ?config(sasl, Config)},
543551
{ok, Connection} = amqp10_client:open_connection(Conf),
544552
{ok, Session} = amqp10_client:begin_session(Connection),
553+
QName = <<"test">>,
554+
Address = declare_queue(Config, QName),
545555
Data = list_to_binary(string:chars(64, 1000)),
546556
{ok, Sender} = amqp10_client:attach_sender_link_sync(Session,
547557
<<"data-sender">>,
548-
<<"test">>),
558+
Address),
549559
Msg = amqp10_msg:new(<<"my-tag">>, Data, true),
550560
ok = amqp10_client:send_msg(Sender, Msg),
551561
{ok, Receiver} = amqp10_client:attach_receiver_link(Session,
552562
<<"data-receiver">>,
553-
<<"test">>),
563+
Address),
554564
{ok, OutMsg} = amqp10_client:get_msg(Receiver),
555565
ok = amqp10_client:end_session(Session),
556566
ok = amqp10_client:close_connection(Connection),
@@ -563,10 +573,12 @@ transfer_unsettled(Config) ->
563573
sasl => ?config(sasl, Config)},
564574
{ok, Connection} = amqp10_client:open_connection(Conf),
565575
{ok, Session} = amqp10_client:begin_session(Connection),
576+
QName = <<"test">>,
577+
Address = declare_queue(Config, QName),
566578
Data = list_to_binary(string:chars(64, 1000)),
567579
{ok, Sender} = amqp10_client:attach_sender_link_sync(Session,
568580
<<"data-sender">>,
569-
<<"test">>, unsettled),
581+
Address, unsettled),
570582
await_link(Sender, credited, credited_timeout),
571583
DeliveryTag = <<"my-tag">>,
572584
Msg = amqp10_msg:new(DeliveryTag, Data, false),
@@ -575,7 +587,7 @@ transfer_unsettled(Config) ->
575587
ok = await_disposition(DeliveryTag),
576588
{ok, Receiver} = amqp10_client:attach_receiver_link(Session,
577589
<<"data-receiver">>,
578-
<<"test">>, unsettled),
590+
Address, unsettled),
579591
{ok, OutMsg} = amqp10_client:get_msg(Receiver),
580592
ok = amqp10_client:accept_msg(Receiver, OutMsg),
581593
{error, timeout} = amqp10_client:get_msg(Receiver, 1000),
@@ -586,17 +598,18 @@ transfer_unsettled(Config) ->
586598
subscribe(Config) ->
587599
Hostname = ?config(rmq_hostname, Config),
588600
Port = rabbit_ct_broker_helpers:get_node_config(Config, 0, tcp_port_amqp),
589-
QueueName = atom_to_binary(?FUNCTION_NAME),
590601
{ok, Connection} = amqp10_client:open_connection(Hostname, Port),
591602
{ok, Session} = amqp10_client:begin_session(Connection),
603+
QName = atom_to_binary(?FUNCTION_NAME),
604+
Address = declare_queue(Config, QName),
592605
{ok, Sender} = amqp10_client:attach_sender_link_sync(Session,
593606
<<"sub-sender">>,
594-
QueueName),
607+
Address),
595608
await_link(Sender, credited, link_credit_timeout),
596609
_ = publish_messages(Sender, <<"banana">>, 10),
597610
{ok, Receiver} = amqp10_client:attach_receiver_link(Session,
598611
<<"sub-receiver">>,
599-
QueueName, unsettled),
612+
Address, unsettled),
600613
ok = amqp10_client:flow_link_credit(Receiver, 10, never),
601614
[begin
602615
receive {amqp10_msg, Receiver, Msg} ->
@@ -618,17 +631,18 @@ subscribe_with_auto_flow_settled(Config) ->
618631
SenderSettleMode = settled,
619632
Hostname = ?config(rmq_hostname, Config),
620633
Port = rabbit_ct_broker_helpers:get_node_config(Config, 0, tcp_port_amqp),
621-
QueueName = atom_to_binary(?FUNCTION_NAME),
622634
{ok, Connection} = amqp10_client:open_connection(Hostname, Port),
623635
{ok, Session} = amqp10_client:begin_session(Connection),
636+
QName = atom_to_binary(?FUNCTION_NAME),
637+
Address = declare_queue(Config, QName),
624638
{ok, Sender} = amqp10_client:attach_sender_link_sync(Session,
625639
<<"sub-sender">>,
626-
QueueName),
640+
Address),
627641
await_link(Sender, credited, link_credit_timeout),
628642

629643
publish_messages(Sender, <<"banana">>, 20),
630644
{ok, Receiver} = amqp10_client:attach_receiver_link(
631-
Session, <<"sub-receiver">>, QueueName, SenderSettleMode),
645+
Session, <<"sub-receiver">>, Address, SenderSettleMode),
632646
await_link(Receiver, attached, attached_timeout),
633647

634648
ok = amqp10_client:flow_link_credit(Receiver, 5, 2),
@@ -643,18 +657,19 @@ subscribe_with_auto_flow_unsettled(Config) ->
643657
SenderSettleMode = unsettled,
644658
Hostname = ?config(rmq_hostname, Config),
645659
Port = rabbit_ct_broker_helpers:get_node_config(Config, 0, tcp_port_amqp),
646-
QueueName = atom_to_binary(?FUNCTION_NAME),
647660
{ok, Connection} = amqp10_client:open_connection(Hostname, Port),
648661
{ok, Session} = amqp10_client:begin_session(Connection),
662+
QName = atom_to_binary(?FUNCTION_NAME),
663+
Address = declare_queue(Config, QName),
649664
{ok, Sender} = amqp10_client:attach_sender_link_sync(Session,
650665
<<"sub-sender">>,
651-
QueueName),
666+
Address),
652667
await_link(Sender, credited, link_credit_timeout),
653668

654669
_ = publish_messages(Sender, <<"1-">>, 30),
655670
%% Use sender settle mode 'unsettled'.
656671
%% This should require us to manually settle message in order to receive more messages.
657-
{ok, Receiver} = amqp10_client:attach_receiver_link(Session, <<"sub-receiver-2">>, QueueName, SenderSettleMode),
672+
{ok, Receiver} = amqp10_client:attach_receiver_link(Session, <<"sub-receiver-2">>, Address, SenderSettleMode),
658673
await_link(Receiver, attached, attached_timeout),
659674
ok = amqp10_client:flow_link_credit(Receiver, 5, 2),
660675
%% We should receive exactly 5 messages.
@@ -966,6 +981,22 @@ incoming_heartbeat(Config) ->
966981
%%% HELPERS
967982
%%%
968983

984+
declare_queue(Config, QName) ->
985+
case ?config(name, ?config(tc_group_properties, Config)) of
986+
activemq ->
987+
ok;
988+
_ ->
989+
Q = rabbit_misc:r(<<"/">>, queue, QName),
990+
{_, _} = rabbit_ct_broker_helpers:rpc(Config, rabbit_amqqueue, declare,
991+
[Q, true, false, [], none, <<"acting-user">>])
992+
end,
993+
queue_address(QName).
994+
995+
%% From rabbitmq_amqp_address.
996+
queue_address(QueueName) ->
997+
QueueNameQuoted = uri_string:quote(QueueName),
998+
<<"/queues/", QueueNameQuoted/binary>>.
999+
9691000
await_link(Who, What, Err) ->
9701001
receive
9711002
{amqp10_event, {link, Who0, What0}}

‎deps/rabbit/src/rabbit_amqp_session.erl‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,7 @@
2020

2121
-rabbit_deprecated_feature(
2222
{amqp_address_v1,
23-
#{deprecation_phase => permitted_by_default,
23+
#{deprecation_phase => denied_by_default,
2424
doc_url => "https://www.rabbitmq.com/docs/amqp#addresses",
2525
messages =>
2626
#{when_permitted =>

‎deps/rabbit/test/amqp_client_SUITE.erl‎

Lines changed: 17 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -1741,6 +1741,8 @@ amqp091_to_amqp10_header_conversion(Session, Ch, QName, Address) ->
17411741

17421742
%% Test sending and receiving concurrently on multiple sessions of the same connection.
17431743
multiple_sessions(Config) ->
1744+
{Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config),
1745+
17441746
OpnConf = connection_config(Config),
17451747
{ok, Connection} = amqp10_client:open_connection(OpnConf),
17461748
%% Create 2 sessions on the same connection.
@@ -1751,10 +1753,18 @@ multiple_sessions(Config) ->
17511753
Q1 = <<"q1">>,
17521754
Q2 = <<"q2">>,
17531755
Qs = [Q1, Q2],
1756+
#'queue.declare_ok'{} = amqp_channel:call(
1757+
Ch, #'queue.declare'{
1758+
queue = Q1,
1759+
durable = true}),
1760+
#'queue.declare_ok'{} = amqp_channel:call(
1761+
Ch, #'queue.declare'{
1762+
queue = Q2,
1763+
durable = true}),
17541764
{ok, Receiver1} = amqp10_client:attach_receiver_link(
1755-
Session1, <<"receiver link 1">>, Q1, settled, configuration),
1765+
Session1, <<"receiver link 1">>, rabbitmq_amqp_address:queue(Q1), settled, configuration),
17561766
{ok, Receiver2} = amqp10_client:attach_receiver_link(
1757-
Session2, <<"receiver link 2">>, Q2, settled, configuration),
1767+
Session2, <<"receiver link 2">>, rabbitmq_amqp_address:queue(Q2), settled, configuration),
17581768
receive {amqp10_event, {link, Receiver1, attached}} -> ok
17591769
after 30000 -> ct:fail("missing attached")
17601770
end,
@@ -1767,7 +1777,6 @@ multiple_sessions(Config) ->
17671777
ok = amqp10_client:flow_link_credit(Receiver2, NMsgsPerReceiver, never),
17681778
flush("receiver attached"),
17691779

1770-
{Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config),
17711780
[#'queue.bind_ok'{} = amqp_channel:call(Ch, #'queue.bind'{queue = QName,
17721781
exchange = <<"amq.fanout">>})
17731782
|| QName <- Qs],
@@ -3203,7 +3212,7 @@ detach_requeues_two_connections_quorum_queue(Config) ->
32033212

32043213
detach_requeues_two_connections(QType, Config) ->
32053214
QName = atom_to_binary(?FUNCTION_NAME),
3206-
Address = <<"/queue/", QName/binary>>,
3215+
Address = rabbitmq_amqp_address:queue(QName),
32073216

32083217
%% Connect to new node.
32093218
OpnConf0 = connection_config(0, Config),
@@ -4065,7 +4074,7 @@ queue_and_client_different_nodes(QueueLeaderNode, ClientNode, QueueType, Config)
40654074
OpnConf = connection_config(ClientNode, Config),
40664075
{ok, Connection} = amqp10_client:open_connection(OpnConf),
40674076
{ok, Session} = amqp10_client:begin_session_sync(Connection),
4068-
Address = <<"/amq/queue/", QName/binary>>,
4077+
Address = rabbitmq_amqp_address:queue(QName),
40694078
{ok, Sender} = amqp10_client:attach_sender_link(
40704079
Session, <<"test-sender">>, Address),
40714080
ok = wait_for_credit(Sender),
@@ -5759,7 +5768,7 @@ dead_letter_into_stream(Config) ->
57595768
}}),
57605769
60_000, 5000),
57615770
{ok, Receiver} = amqp10_client:attach_receiver_link(
5762-
Session1, <<"receiver">>, <<"/amq/queue/", QName1/binary>>,
5771+
Session1, <<"receiver">>, rabbitmq_amqp_address:queue(QName1),
57635772
settled, configuration,
57645773
#{<<"rabbitmq:stream-offset-spec">> => <<"first">>}),
57655774
{ok, Sender} = amqp10_client:attach_sender_link(
@@ -5835,7 +5844,7 @@ accept_multiple_message_order(QType, Config) ->
58355844
ok = amqp10_client:detach_link(Sender),
58365845
assert_messages(QName, 5, 0, Config),
58375846

5838-
{ok, Receiver} = amqp10_client:attach_receiver_link(Session, <<"receiver">>, QName, unsettled),
5847+
{ok, Receiver} = amqp10_client:attach_receiver_link(Session, <<"receiver">>, Address, unsettled),
58395848
{ok, Msg1} = amqp10_client:get_msg(Receiver),
58405849
{ok, Msg2} = amqp10_client:get_msg(Receiver),
58415850
{ok, _Msg3} = amqp10_client:get_msg(Receiver),
@@ -5886,7 +5895,7 @@ release_multiple_message_order(QType, Config) ->
58865895
ok = amqp10_client:detach_link(Sender),
58875896
assert_messages(QName, 4, 0, Config),
58885897

5889-
{ok, Receiver} = amqp10_client:attach_receiver_link(Session, <<"receiver">>, QName, unsettled),
5898+
{ok, Receiver} = amqp10_client:attach_receiver_link(Session, <<"receiver">>, Address, unsettled),
58905899
{ok, Msg1} = amqp10_client:get_msg(Receiver),
58915900
{ok, Msg2} = amqp10_client:get_msg(Receiver),
58925901
{ok, Msg3} = amqp10_client:get_msg(Receiver),

0 commit comments

Comments
 (0)