root
/
rabbitmq-server
mirror of https://github.com/rabbitmq/rabbitmq-server.git
1
0
Fork 0
Code Issues Actions 1 Packages Projects Releases Wiki Activity

Merge pull request #11412 from rabbitmq/link-error 

Prefer link error over session error
This commit is contained in:
Michael Klishin 2024-06-07 11:10:48 -04:00 committed by GitHub
parent 5e4a4326ef 895bf3e3cb
commit b9f387b988
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 64 additions and 55 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

77
deps/rabbit/src/rabbit_amqp_session.erl vendored
View File

@ -2305,12 +2305,8 @@ incoming_link_transfer(
"delivery_tag=~p, delivery_id=~p, reason=~p", "delivery_tag=~p, delivery_id=~p, reason=~p",
[DeliveryTag, DeliveryId, Reason]) [DeliveryTag, DeliveryId, Reason])
end; end;
{error, not_found, XName} -> {error, #'v1_0.error'{} = Err} ->
Disposition = released(DeliveryId), Disposition = released(DeliveryId),
Description = unicode:characters_to_binary("no " ++ rabbit_misc:rs(XName)),
Err = #'v1_0.error'{
condition = ?V_1_0_AMQP_ERROR_NOT_FOUND,
description = {utf8, Description}},
Detach = detach(HandleInt, Link0, Err), Detach = detach(HandleInt, Link0, Err),
{error, [Disposition, Detach]} {error, [Disposition, Detach]}
end. end.
@ -2322,7 +2318,7 @@ lookup_target(#resource{} = XName, LinkRKey, Mc, _, _, PermCache) ->
{ok, X} -> {ok, X} ->
lookup_routing_key(X, LinkRKey, Mc, PermCache); lookup_routing_key(X, LinkRKey, Mc, PermCache);
{error, not_found} -> {error, not_found} ->
{error, not_found, XName} {error, error_not_found(XName)}
end; end;
lookup_target(to, to, Mc, Vhost, User, PermCache0) -> lookup_target(to, to, Mc, Vhost, User, PermCache0) ->
case mc:property(to, Mc) of case mc:property(to, Mc) of
@ -2336,19 +2332,19 @@ lookup_target(to, to, Mc, Vhost, User, PermCache0) ->
check_internal_exchange(X), check_internal_exchange(X),
lookup_routing_key(X, RKey, Mc, PermCache); lookup_routing_key(X, RKey, Mc, PermCache);
{error, not_found} -> {error, not_found} ->
{error, not_found, XName} {error, error_not_found(XName)}
end; end;
{error, bad_address} -> {error, bad_address} ->
protocol_error( {error,
?V_1_0_AMQP_ERROR_PRECONDITION_FAILED, #'v1_0.error'{
"bad 'to' address string: ~ts", condition = ?V_1_0_AMQP_ERROR_PRECONDITION_FAILED,
[String]) description = {utf8, <<"bad 'to' address string: ", String/binary>>}}}
end; end;
undefined -> undefined ->
protocol_error( {error,
?V_1_0_AMQP_ERROR_PRECONDITION_FAILED, #'v1_0.error'{
"anonymous terminus requires 'to' address to be set", condition = ?V_1_0_AMQP_ERROR_PRECONDITION_FAILED,
[]) description = {utf8, <<"anonymous terminus requires 'to' address to be set">>}}}
end. end.
lookup_routing_key(X = #exchange{name = #resource{name = XNameBin}}, lookup_routing_key(X = #exchange{name = #resource{name = XNameBin}},
@ -2430,7 +2426,7 @@ maybe_grant_mgmt_link_credit(Credit, _, _) ->
{ok, rabbit_amqqueue:name(), permission_cache(), topic_permission_cache()} | {ok, rabbit_amqqueue:name(), permission_cache(), topic_permission_cache()} |
{error, term()}. {error, term()}.
ensure_source(#'v1_0.source'{dynamic = true}, _, _, _, _) -> ensure_source(#'v1_0.source'{dynamic = true}, _, _, _, _) ->
not_implemented("Dynamic sources not supported"); exit_not_implemented("Dynamic sources not supported");
ensure_source(#'v1_0.source'{address = Address, ensure_source(#'v1_0.source'{address = Address,
durable = Durable}, durable = Durable},
Vhost, User, PermCache, TopicPermCache) -> Vhost, User, PermCache, TopicPermCache) ->
@ -2504,7 +2500,7 @@ ensure_source_v2(Address, _, _, _) ->
permission_cache()} | permission_cache()} |
{error, term()}. {error, term()}.
ensure_target(#'v1_0.target'{dynamic = true}, _, _, _) -> ensure_target(#'v1_0.target'{dynamic = true}, _, _, _) ->
not_implemented("Dynamic targets not supported"); exit_not_implemented("Dynamic targets not supported");
ensure_target(#'v1_0.target'{address = Address, ensure_target(#'v1_0.target'{address = Address,
durable = Durable}, durable = Durable},
Vhost, User, PermCache) -> Vhost, User, PermCache) ->
@ -2549,7 +2545,7 @@ check_exchange(XNameBin, RKey, QNameBin, User, Vhost, PermCache0) ->
end, end,
{ok, Exchange, RKey, QNameBin, PermCache}; {ok, Exchange, RKey, QNameBin, PermCache};
{error, not_found} -> {error, not_found} ->
not_found(XName) exit_not_found(XName)
end. end.
ensure_target_v1({utf8, Address}, Vhost, User, Durable, PermCache0) -> ensure_target_v1({utf8, Address}, Vhost, User, Durable, PermCache0) ->
@ -2908,16 +2904,16 @@ keyfind_unpack_described(Key, KvList) ->
end. end.
validate_attach(#'v1_0.attach'{target = #'v1_0.coordinator'{}}) -> validate_attach(#'v1_0.attach'{target = #'v1_0.coordinator'{}}) ->
not_implemented("Transactions not supported"); exit_not_implemented("Transactions not supported");
validate_attach(#'v1_0.attach'{unsettled = {map, [_|_]}}) -> validate_attach(#'v1_0.attach'{unsettled = {map, [_|_]}}) ->
not_implemented("Link recovery not supported"); exit_not_implemented("Link recovery not supported");
validate_attach(#'v1_0.attach'{incomplete_unsettled = true}) -> validate_attach(#'v1_0.attach'{incomplete_unsettled = true}) ->
not_implemented("Link recovery not supported"); exit_not_implemented("Link recovery not supported");
validate_attach( validate_attach(
#'v1_0.attach'{snd_settle_mode = SndSettleMode, #'v1_0.attach'{snd_settle_mode = SndSettleMode,
rcv_settle_mode = ?V_1_0_RECEIVER_SETTLE_MODE_SECOND}) rcv_settle_mode = ?V_1_0_RECEIVER_SETTLE_MODE_SECOND})
when SndSettleMode =/= ?V_1_0_SENDER_SETTLE_MODE_SETTLED -> when SndSettleMode =/= ?V_1_0_SENDER_SETTLE_MODE_SETTLED ->
not_implemented("rcv-settle-mode second not supported"); exit_not_implemented("rcv-settle-mode second not supported");
validate_attach(#'v1_0.attach'{}) -> validate_attach(#'v1_0.attach'{}) ->
ok. ok.
@ -2951,7 +2947,7 @@ validate_multi_transfer_settled(Other, First)
%% "If the message is being sent settled by the sender, %% "If the message is being sent settled by the sender,
%% the value of this field [rcv-settle-mode] is ignored." [2.7.5] %% the value of this field [rcv-settle-mode] is ignored." [2.7.5]
validate_transfer_rcv_settle_mode(?V_1_0_RECEIVER_SETTLE_MODE_SECOND, _Settled = false) -> validate_transfer_rcv_settle_mode(?V_1_0_RECEIVER_SETTLE_MODE_SECOND, _Settled = false) ->
not_implemented("rcv-settle-mode second not supported"); exit_not_implemented("rcv-settle-mode second not supported");
validate_transfer_rcv_settle_mode(_, _) -> validate_transfer_rcv_settle_mode(_, _) ->
ok. ok.
@ -3025,7 +3021,7 @@ exit_if_absent(ResourceName = #resource{kind = Kind}) ->
end, end,
case Mod:exists(ResourceName) of case Mod:exists(ResourceName) of
true -> ok; true -> ok;
false -> not_found(ResourceName) false -> exit_not_found(ResourceName)
end. end.
generate_queue_name() -> generate_queue_name() ->
@ -3072,10 +3068,10 @@ outcomes(#'v1_0.source'{outcomes = {array, symbol, Syms} = Outcomes}) ->
[] -> [] ->
Outcomes; Outcomes;
Unsupported -> Unsupported ->
not_implemented("Outcomes not supported: ~tp", [Unsupported]) exit_not_implemented("Outcomes not supported: ~tp", [Unsupported])
end; end;
outcomes(#'v1_0.source'{outcomes = Unsupported}) -> outcomes(#'v1_0.source'{outcomes = Unsupported}) ->
not_implemented("Outcomes not supported: ~tp", [Unsupported]); exit_not_implemented("Outcomes not supported: ~tp", [Unsupported]);
outcomes(_) -> outcomes(_) ->
{array, symbol, ?OUTCOMES}. {array, symbol, ?OUTCOMES}.
@ -3330,30 +3326,37 @@ check_paired({map, Properties}) ->
true -> true ->
ok; ok;
false -> false ->
property_paired_not_set() exit_property_paired_not_set()
end; end;
check_paired(_) -> check_paired(_) ->
property_paired_not_set(). exit_property_paired_not_set().
-spec property_paired_not_set() -> no_return(). -spec exit_property_paired_not_set() -> no_return().
property_paired_not_set() -> exit_property_paired_not_set() ->
protocol_error(?V_1_0_AMQP_ERROR_INVALID_FIELD, protocol_error(?V_1_0_AMQP_ERROR_INVALID_FIELD,
"Link property 'paired' is not set to boolean value 'true'", []). "Link property 'paired' is not set to boolean value 'true'", []).
-spec not_implemented(io:format()) -> no_return(). -spec exit_not_implemented(io:format()) -> no_return().
not_implemented(Format) -> exit_not_implemented(Format) ->
not_implemented(Format, []). exit_not_implemented(Format, []).
-spec not_implemented(io:format(), [term()]) -> no_return(). -spec exit_not_implemented(io:format(), [term()]) -> no_return().
not_implemented(Format, Args) -> exit_not_implemented(Format, Args) ->
protocol_error(?V_1_0_AMQP_ERROR_NOT_IMPLEMENTED, Format, Args). protocol_error(?V_1_0_AMQP_ERROR_NOT_IMPLEMENTED, Format, Args).
-spec not_found(rabbit_types:r(exchange | queue)) -> no_return(). -spec exit_not_found(rabbit_types:r(exchange | queue)) -> no_return().
not_found(Resource) -> exit_not_found(Resource) ->
protocol_error(?V_1_0_AMQP_ERROR_NOT_FOUND, protocol_error(?V_1_0_AMQP_ERROR_NOT_FOUND,
"no ~ts", "no ~ts",
[rabbit_misc:rs(Resource)]). [rabbit_misc:rs(Resource)]).
-spec error_not_found(rabbit_types:r(exchange | queue)) -> #'v1_0.error'{}.
error_not_found(Resource) ->
Description = unicode:characters_to_binary("no " ++ rabbit_misc:rs(Resource)),
#'v1_0.error'{
condition = ?V_1_0_AMQP_ERROR_NOT_FOUND,
description = {utf8, Description}}.
address_v1_permitted() -> address_v1_permitted() ->
rabbit_deprecated_features:is_permitted(amqp_address_v1). rabbit_deprecated_features:is_permitted(amqp_address_v1).

42
deps/rabbit/test/amqp_address_SUITE.erl vendored
View File

@ -391,16 +391,19 @@ target_per_message_unset_to_address(Config) ->
ok = wait_for_credit(Sender), ok = wait_for_credit(Sender),
%% Send message with 'to' unset. %% Send message with 'to' unset.
ok = amqp10_client:send_msg(Sender, amqp10_msg:new(<<0>>, <<0>>)), DTag = <<1>>,
receive ok = amqp10_client:send_msg(Sender, amqp10_msg:new(DTag, <<0>>)),
{amqp10_event, ok = wait_for_settled(released, DTag),
{session, Session, receive {amqp10_event,
{ended, {link, Sender,
#'v1_0.error'{ {detached,
condition = ?V_1_0_AMQP_ERROR_PRECONDITION_FAILED, #'v1_0.error'{
description = {utf8, <<"anonymous terminus requires 'to' address to be set">>}}}}} -> ok condition = ?V_1_0_AMQP_ERROR_PRECONDITION_FAILED,
after 5000 -> ct:fail({missing_event, ?LINE}) description = {utf8, <<"anonymous terminus requires 'to' address to be set">>}}}}} -> ok
after 5000 -> ct:fail("server did not close our outgoing link")
end, end,
ok = amqp10_client:end_session(Session),
ok = amqp10_client:close_connection(Connection). ok = amqp10_client:close_connection(Connection).
bad_v2_addresses() -> bad_v2_addresses() ->
@ -436,17 +439,20 @@ target_per_message_bad_to_address0(Address, Config) ->
{ok, Sender} = amqp10_client:attach_sender_link(Session, <<"sender">>, null), {ok, Sender} = amqp10_client:attach_sender_link(Session, <<"sender">>, null),
ok = wait_for_credit(Sender), ok = wait_for_credit(Sender),
Msg = amqp10_msg:set_properties(#{to => Address}, amqp10_msg:new(<<0>>, <<0>>)), DTag = <<255>>,
Msg = amqp10_msg:set_properties(#{to => Address}, amqp10_msg:new(DTag, <<0>>)),
ok = amqp10_client:send_msg(Sender, Msg), ok = amqp10_client:send_msg(Sender, Msg),
receive ok = wait_for_settled(released, DTag),
{amqp10_event, receive {amqp10_event,
{session, Session, {link, Sender,
{ended, {detached,
#'v1_0.error'{ #'v1_0.error'{
condition = ?V_1_0_AMQP_ERROR_PRECONDITION_FAILED, condition = ?V_1_0_AMQP_ERROR_PRECONDITION_FAILED,
description = {utf8, <<"bad 'to' address", _Rest/binary>>}}}}} -> ok description = {utf8, <<"bad 'to' address", _Rest/binary>>}}}}} -> ok
after 5000 -> ct:fail({missing_event, ?LINE, Address}) after 5000 -> ct:fail("server did not close our outgoing link")
end, end,
ok = amqp10_client:end_session(Session),
ok = amqp10_client:close_connection(Connection). ok = amqp10_client:close_connection(Connection).
target_per_message_exchange_absent(Config) -> target_per_message_exchange_absent(Config) ->