Skip to content

Commit 81caabc

Browse files
ansdansd
authored
Merge pull request #14389 from rabbitmq/refuse-link
Detach link for link-level errors
2 parents 1fc7390 + eef470d commit 81caabc

File tree

8 files changed

+475
-264
lines changed

8 files changed

+475
-264
lines changed

‎deps/amqp10_client/src/amqp10_client_session.erl

Lines changed: 53 additions & 48 deletions
Original file line numberDiff line numberDiff line change
@@ -127,7 +127,7 @@
127127
-record(link,
128128
{name :: link_name(),
129129
ref :: link_ref(),
130-
state = detached :: detached | attach_sent | attached | detach_sent,
130+
state = detached :: detached | attach_sent | attached | attach_refused | detach_sent,
131131
notify :: pid(),
132132
output_handle :: output_handle(),
133133
input_handle :: input_handle() | undefined,
@@ -325,9 +325,11 @@ mapped(cast, #'v1_0.end'{} = End, State) ->
325325
ok = notify_session_ended(End, State),
326326
{stop, normal, State};
327327
mapped(cast, #'v1_0.attach'{name = {utf8, Name},
328-
initial_delivery_count = IDC,
329328
handle = {uint, InHandle},
330329
role = PeerRoleBool,
330+
source = Source,
331+
target = Target,
332+
initial_delivery_count = IDC,
331333
max_message_size = MaybeMaxMessageSize} = Attach,
332334
#state{links = Links, link_index = LinkIndex,
333335
link_handle_index = LHI} = State0) ->
@@ -339,20 +341,28 @@ mapped(cast, #'v1_0.attach'{name = {utf8, Name},
339341
#{OutHandle := Link0} = Links,
340342
ok = notify_link_attached(Link0, Attach, State0),
341343

342-
{DeliveryCount, MaxMessageSize} =
344+
{LinkState, DeliveryCount, MaxMessageSize} =
343345
case Link0 of
344346
#link{role = sender = OurRole,
345347
delivery_count = DC} ->
348+
LS = case Target of
349+
#'v1_0.target'{} -> attached;
350+
_ -> attach_refused
351+
end,
346352
MSS = case MaybeMaxMessageSize of
347353
{ulong, S} when S > 0 -> S;
348354
_ -> undefined
349355
end,
350-
{DC, MSS};
356+
{LS, DC, MSS};
351357
#link{role = receiver = OurRole,
352358
max_message_size = MSS} ->
353-
{unpack(IDC), MSS}
359+
LS = case Source of
360+
#'v1_0.source'{} -> attached;
361+
_ -> attach_refused
362+
end,
363+
{LS, unpack(IDC), MSS}
354364
end,
355-
Link = Link0#link{state = attached,
365+
Link = Link0#link{state = LinkState,
356366
input_handle = InHandle,
357367
delivery_count = DeliveryCount,
358368
max_message_size = MaxMessageSize},
@@ -496,43 +506,31 @@ mapped({call, From},
496506
{keep_state_and_data, {reply, From, {error, remote_incoming_window_exceeded}}};
497507
mapped({call, From = {Pid, _}},
498508
{transfer, #'v1_0.transfer'{handle = {uint, OutHandle},
499-
delivery_tag = {binary, DeliveryTag},
500-
settled = false} = Transfer0, Sections},
501-
#state{outgoing_delivery_id = DeliveryId, links = Links,
502-
outgoing_unsettled = Unsettled} = State) ->
509+
delivery_tag = DeliveryTag,
510+
settled = Settled} = Transfer0, Sections},
511+
#state{outgoing_delivery_id = DeliveryId,
512+
links = Links,
513+
outgoing_unsettled = Unsettled0} = State0) ->
503514
case Links of
515+
#{OutHandle := #link{state = attach_refused}} ->
516+
{keep_state_and_data, {reply, From, {error, attach_refused}}};
504517
#{OutHandle := #link{input_handle = undefined}} ->
505518
{keep_state_and_data, {reply, From, {error, half_attached}}};
506519
#{OutHandle := #link{link_credit = LC}} when LC =< 0 ->
507520
{keep_state_and_data, {reply, From, {error, insufficient_credit}}};
508-
#{OutHandle := Link = #link{max_message_size = MaxMessageSize,
509-
footer_opt = FooterOpt}} ->
521+
#{OutHandle := #link{max_message_size = MaxMessageSize,
522+
footer_opt = FooterOpt} = Link} ->
510523
Transfer = Transfer0#'v1_0.transfer'{delivery_id = uint(DeliveryId)},
511-
case send_transfer(Transfer, Sections, FooterOpt, MaxMessageSize, State) of
512-
{ok, NumFrames} ->
513-
State1 = State#state{outgoing_unsettled = Unsettled#{DeliveryId => {DeliveryTag, Pid}}},
514-
{keep_state, book_transfer_send(NumFrames, Link, State1), {reply, From, ok}};
515-
Error ->
516-
{keep_state_and_data, {reply, From, Error}}
517-
end;
518-
_ ->
519-
{keep_state_and_data, {reply, From, {error, link_not_found}}}
520-
521-
end;
522-
mapped({call, From},
523-
{transfer, #'v1_0.transfer'{handle = {uint, OutHandle}} = Transfer0,
524-
Sections}, #state{outgoing_delivery_id = DeliveryId,
525-
links = Links} = State) ->
526-
case Links of
527-
#{OutHandle := #link{input_handle = undefined}} ->
528-
{keep_state_and_data, {reply, From, {error, half_attached}}};
529-
#{OutHandle := #link{link_credit = LC}} when LC =< 0 ->
530-
{keep_state_and_data, {reply, From, {error, insufficient_credit}}};
531-
#{OutHandle := Link = #link{max_message_size = MaxMessageSize,
532-
footer_opt = FooterOpt}} ->
533-
Transfer = Transfer0#'v1_0.transfer'{delivery_id = uint(DeliveryId)},
534-
case send_transfer(Transfer, Sections, FooterOpt, MaxMessageSize, State) of
524+
case send_transfer(Transfer, Sections, FooterOpt, MaxMessageSize, State0) of
535525
{ok, NumFrames} ->
526+
State = case Settled of
527+
true ->
528+
State0;
529+
false ->
530+
{binary, Tag} = DeliveryTag,
531+
Unsettled = Unsettled0#{DeliveryId => {Tag, Pid}},
532+
State0#state{outgoing_unsettled = Unsettled}
533+
end,
536534
{keep_state, book_transfer_send(NumFrames, Link, State), {reply, From, ok}};
537535
Error ->
538536
{keep_state_and_data, {reply, From, Error}}
@@ -688,21 +686,28 @@ send_flow_link(OutHandle,
688686
never -> never;
689687
_ -> {RenewWhenBelow, Credit}
690688
end,
691-
#{OutHandle := #link{output_handle = H,
689+
#{OutHandle := #link{state = LinkState,
690+
output_handle = H,
692691
role = receiver,
693692
delivery_count = DeliveryCount,
694693
available = Available} = Link} = Links,
695-
Flow1 = Flow0#'v1_0.flow'{
696-
handle = uint(H),
697-
%% "In the event that the receiving link endpoint has not yet seen the
698-
%% initial attach frame from the sender this field MUST NOT be set." [2.7.4]
699-
delivery_count = maybe_uint(DeliveryCount),
700-
available = uint(Available)},
701-
Flow = set_flow_session_fields(Flow1, State),
702-
ok = send(Flow, State),
703-
State#state{links = Links#{OutHandle =>
704-
Link#link{link_credit = Credit,
705-
auto_flow = AutoFlow}}}.
694+
case LinkState of
695+
attach_refused ->
696+
%% We will receive the DETACH frame shortly.
697+
State;
698+
_ ->
699+
Flow1 = Flow0#'v1_0.flow'{
700+
handle = uint(H),
701+
%% "In the event that the receiving link endpoint has not yet seen the
702+
%% initial attach frame from the sender this field MUST NOT be set." [2.7.4]
703+
delivery_count = maybe_uint(DeliveryCount),
704+
available = uint(Available)},
705+
Flow = set_flow_session_fields(Flow1, State),
706+
ok = send(Flow, State),
707+
State#state{links = Links#{OutHandle =>
708+
Link#link{link_credit = Credit,
709+
auto_flow = AutoFlow}}}
710+
end.
706711

707712
send_flow_session(State) ->
708713
Flow = set_flow_session_fields(#'v1_0.flow'{}, State),

‎deps/amqp10_client/test/system_SUITE.erl

Lines changed: 52 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -52,6 +52,7 @@ groups() ->
5252
]},
5353
{mock, [], [
5454
insufficient_credit,
55+
attach_refused,
5556
incoming_heartbeat,
5657
multi_transfer_without_delivery_id
5758
]}
@@ -772,11 +773,13 @@ insufficient_credit(Config) ->
772773
outgoing_window = {uint, 1000}}
773774
]}
774775
end,
775-
AttachStep = fun({0 = Ch, #'v1_0.attach'{role = false,
776-
name = Name}, <<>>}) ->
776+
AttachStep = fun({0 = Ch, #'v1_0.attach'{name = Name,
777+
role = false,
778+
target = Target}, <<>>}) ->
777779
{Ch, [#'v1_0.attach'{name = Name,
778780
handle = {uint, 99},
779-
role = true}]}
781+
role = true,
782+
target = Target}]}
780783
end,
781784
Steps = [fun mock_server:recv_amqp_header_step/1,
782785
fun mock_server:send_amqp_header_step/1,
@@ -799,6 +802,52 @@ insufficient_credit(Config) ->
799802
ok = amqp10_client:close_connection(Connection),
800803
ok.
801804

805+
attach_refused(Config) ->
806+
Hostname = ?config(mock_host, Config),
807+
Port = ?config(mock_port, Config),
808+
OpenStep = fun({0 = Ch, #'v1_0.open'{}, _Pay}) ->
809+
{Ch, [#'v1_0.open'{container_id = {utf8, <<"mock">>}}]}
810+
end,
811+
BeginStep = fun({0 = Ch, #'v1_0.begin'{}, _Pay}) ->
812+
{Ch, [#'v1_0.begin'{remote_channel = {ushort, Ch},
813+
next_outgoing_id = {uint, 1},
814+
incoming_window = {uint, 1000},
815+
outgoing_window = {uint, 1000}}
816+
]}
817+
end,
818+
AttachStep = fun({0 = Ch, #'v1_0.attach'{name = Name,
819+
role = false}, <<>>}) ->
820+
%% We test only the 1st stage of link refusal:
821+
%% Server replies with its local terminus set to null.
822+
%% We omit the 2nd stage (the detach frame).
823+
{Ch, [#'v1_0.attach'{name = Name,
824+
handle = {uint, 99},
825+
role = true,
826+
target = undefined}]}
827+
end,
828+
Steps = [fun mock_server:recv_amqp_header_step/1,
829+
fun mock_server:send_amqp_header_step/1,
830+
mock_server:amqp_step(OpenStep),
831+
mock_server:amqp_step(BeginStep),
832+
mock_server:amqp_step(AttachStep)],
833+
834+
ok = mock_server:set_steps(?config(mock_server, Config), Steps),
835+
836+
Cfg = #{address => Hostname, port => Port, sasl => none, notify => self()},
837+
{ok, Connection} = amqp10_client:open_connection(Cfg),
838+
{ok, Session} = amqp10_client:begin_session_sync(Connection),
839+
{ok, Sender} = amqp10_client:attach_sender_link(Session, <<"mock1-sender">>,
840+
<<"test">>),
841+
await_link(Sender, attached, attached_timeout),
842+
Msg = amqp10_msg:new(<<"mock-tag">>, <<"banana">>, true),
843+
%% We expect that the lib prevents the app from sending messages
844+
%% in this intermediate link refusal state.
845+
?assertEqual({error, attach_refused},
846+
amqp10_client:send_msg(Sender, Msg)),
847+
848+
ok = amqp10_client:end_session(Session),
849+
ok = amqp10_client:close_connection(Connection).
850+
802851
multi_transfer_without_delivery_id(Config) ->
803852
Hostname = ?config(mock_host, Config),
804853
Port = ?config(mock_port, Config),

0 commit comments

Comments
 (0)