more link errors

Author: ansd

deps/rabbit/src/rabbit_amqp_session.erl

@@ -1125,32 +1125,34 @@ handle_frame(#'v1_0.attach'{name = {utf8, NameBin} = Name,
                             source = Source,
                             target = Target} = Attach,
              State) ->
-    ok = validate_attach(Attach),
-    try handle_attach(Attach, State)
-    catch {link_error, Error} ->
-              %% Figure 2.33
-              ?LOG_WARNING("refusing link '~ts': ~tp", [NameBin, Error]),
-              AttachReply = case Role of
-                                ?AMQP_ROLE_SENDER ->
-                                    #'v1_0.attach'{
-                                       name = Name,
-                                       handle = Handle,
-                                       role = ?AMQP_ROLE_RECEIVER,
-                                       source = Source,
-                                       target = null};
-                                ?AMQP_ROLE_RECEIVER ->
-                                    #'v1_0.attach'{
-                                       name = Name,
-                                       handle = Handle,
-                                       role = ?AMQP_ROLE_SENDER,
-                                       source = null,
-                                       target = Target,
-                                       initial_delivery_count = ?UINT(?INITIAL_DELIVERY_COUNT)}
-                            end,
-              Detach = #'v1_0.detach'{handle = Handle,
-                                      closed = true,
-                                      error = Error},
-              {ok, [AttachReply, Detach], State}
+    try
+        ok = validate_attach(Attach),
+        handle_attach(Attach, State)
+    catch
+        {link_error, Error} ->
+            %% Figure 2.33
+            ?LOG_WARNING("refusing link '~ts': ~tp", [NameBin, Error]),
+            AttachReply = case Role of
+                              ?AMQP_ROLE_SENDER ->
+                                  #'v1_0.attach'{
+                                     name = Name,
+                                     handle = Handle,
+                                     role = ?AMQP_ROLE_RECEIVER,
+                                     source = Source,
+                                     target = null};
+                              ?AMQP_ROLE_RECEIVER ->
+                                  #'v1_0.attach'{
+                                     name = Name,
+                                     handle = Handle,
+                                     role = ?AMQP_ROLE_SENDER,
+                                     source = null,
+                                     target = Target,
+                                     initial_delivery_count = ?UINT(?INITIAL_DELIVERY_COUNT)}
+                          end,
+            Detach = #'v1_0.detach'{handle = Handle,
+                                    closed = true,
+                                    error = Error},
+            {ok, [AttachReply, Detach], State}
     end;
 
 handle_frame(Detach = #'v1_0.detach'{handle = ?UINT(HandleInt)},
@@ -1257,9 +1259,9 @@ handle_attach(#'v1_0.attach'{
                                 Pair#management_link_pair{incoming_half = HandleInt},
                                 Pairs0);
                 #{LinkName := Other} ->
-                    protocol_error(?V_1_0_AMQP_ERROR_PRECONDITION_FAILED,
-                                   "received invalid attach ~p for management link pair ~p",
-                                   [Attach, Other]);
+                    link_error(?V_1_0_AMQP_ERROR_PRECONDITION_FAILED,
+                               "received invalid attach ~p for management link pair ~p",
+                               [Attach, Other]);
                 _ ->
                     maps:put(LinkName,
                              #management_link_pair{client_terminus_address = ClientTerminusAddress,
@@ -1314,9 +1316,9 @@ handle_attach(#'v1_0.attach'{
                                 Pair#management_link_pair{outgoing_half = HandleInt},
                                 Pairs0);
                 #{LinkName := Other} ->
-                    protocol_error(?V_1_0_AMQP_ERROR_PRECONDITION_FAILED,
-                                   "received invalid attach ~p for management link pair ~p",
-                                   [Attach, Other]);
+                    link_error(?V_1_0_AMQP_ERROR_PRECONDITION_FAILED,
+                               "received invalid attach ~p for management link pair ~p",
+                               [Attach, Other]);
                 _ ->
                     maps:put(LinkName,
                              #management_link_pair{client_terminus_address = ClientTerminusAddress,
@@ -1546,18 +1548,16 @@ handle_attach(#'v1_0.attach'{role = ?AMQP_ROLE_RECEIVER,
                                    rabbit_global_counters:consumer_created(?PROTOCOL),
                                    {ok, [A], State1};
                                {error, _Type, Reason, Args} ->
-                                   protocol_error(
-                                     ?V_1_0_AMQP_ERROR_INTERNAL_ERROR,
-                                     Reason, Args)
+                                   link_error(?V_1_0_AMQP_ERROR_INTERNAL_ERROR,
+                                              Reason, Args)
                            end
                    end) of
                 {ok, Reply, State} ->
                     reply_frames(Reply, State);
                 {error, Reason} ->
-                    protocol_error(
-                      ?V_1_0_AMQP_ERROR_INTERNAL_ERROR,
-                      "Could not operate on ~s: ~tp",
-                      [rabbit_misc:rs(QName), Reason])
+                    link_error(?V_1_0_AMQP_ERROR_INTERNAL_ERROR,
+                               "Could not operate on ~ts: ~tp",
+                               [rabbit_misc:rs(QName), Reason])
             end
     end.
 
@@ -2531,10 +2531,10 @@ incoming_link_transfer(
                              multi_transfer_msg = undefined},
                     {ok, Reply, Link, State};
                 {error, Reason} ->
-                    protocol_error(?V_1_0_AMQP_ERROR_INTERNAL_ERROR,
-                                   "Failed to deliver message to queues, "
-                                   "delivery_tag=~p, delivery_id=~p, reason=~p",
-                                   [DeliveryTag, DeliveryId, Reason])
+                    link_error(
+                      ?V_1_0_AMQP_ERROR_INTERNAL_ERROR,
+                      "failed to deliver message (delivery-tag=~p, delivery-id=~p): ~tp",
+                      [DeliveryTag, DeliveryId, Reason])
             end;
         {error, {anonymous_terminus, false}, #'v1_0.error'{} = Err} ->
             Disposition = case Settled of
@@ -3364,8 +3364,8 @@ validate_multi_transfer_delivery_id(undefined, _FirstDeliveryId) ->
 validate_multi_transfer_delivery_id(OtherId, FirstDeliveryId) ->
     %% "It is an error if the delivery-id on a continuation transfer
     %% differs from the delivery-id on the first transfer of a delivery."
-    protocol_error(
-      ?V_1_0_CONNECTION_ERROR_FRAMING_ERROR,
+    link_error(
+      ?V_1_0_AMQP_ERROR_INVALID_FIELD,
       "delivery-id of continuation transfer (~p) differs from delivery-id on first transfer (~p)",
       [OtherId, FirstDeliveryId]).
 
@@ -3377,8 +3377,8 @@ validate_multi_transfer_settled(undefined, Settled)
     ok;
 validate_multi_transfer_settled(Other, First)
   when is_boolean(First) ->
-    protocol_error(
-      ?V_1_0_CONNECTION_ERROR_FRAMING_ERROR,
+    link_error(
+      ?V_1_0_AMQP_ERROR_INVALID_FIELD,
       "field 'settled' of continuation transfer (~p) differs from "
       "(interpreted) field 'settled' on first transfer (~p)",
       [Other, First]).
@@ -3394,8 +3394,8 @@ validate_transfer_snd_settle_mode(settled, true) ->
     %% then this field MUST be true on at least one transfer frame for a delivery" [2.7.5]
     ok;
 validate_transfer_snd_settle_mode(SndSettleMode, Settled) ->
-    protocol_error(
-      ?V_1_0_CONNECTION_ERROR_FRAMING_ERROR,
+    link_error(
+      ?V_1_0_AMQP_ERROR_INVALID_FIELD,
       "sender settle mode is '~s' but transfer settled flag is interpreted as being '~s'",
       [SndSettleMode, Settled]).
 
@@ -3523,14 +3523,13 @@ declare_queue(QNameBin,
         {existing, _Q} ->
             ok;
         {error, queue_limit_exceeded, Reason, ReasonArgs} ->
-            link_error(
-              ?V_1_0_AMQP_ERROR_RESOURCE_LIMIT_EXCEEDED,
-              Reason,
-              ReasonArgs);
+            link_error(?V_1_0_AMQP_ERROR_RESOURCE_LIMIT_EXCEEDED,
+                       Reason,
+                       ReasonArgs);
         Other ->
-            protocol_error(?V_1_0_AMQP_ERROR_INTERNAL_ERROR,
-                           "Failed to declare ~s: ~p",
-                           [rabbit_misc:rs(QName), Other])
+            link_error(?V_1_0_AMQP_ERROR_INTERNAL_ERROR,
+                       "Failed to declare ~s: ~p",
+                       [rabbit_misc:rs(QName), Other])
     end,
     {ok, PermCache}.
 
@@ -3836,7 +3835,7 @@ check_user_id(Mc, User) ->
         ok ->
             ok;
         {refused, Reason, Args} ->
-            protocol_error(?V_1_0_AMQP_ERROR_UNAUTHORIZED_ACCESS, Reason, Args)
+            link_error(?V_1_0_AMQP_ERROR_UNAUTHORIZED_ACCESS, Reason, Args)
     end.
 
 maps_update_with(Key, Fun, Init, Map) ->
@@ -3864,23 +3863,23 @@ check_paired({map, Properties}) ->
         true ->
             ok;
         false ->
-            exit_property_paired_not_set()
+            error_property_paired_not_set()
     end;
 check_paired(_) ->
-    exit_property_paired_not_set().
+    error_property_paired_not_set().
 
--spec exit_property_paired_not_set() -> no_return().
-exit_property_paired_not_set() ->
-    protocol_error(?V_1_0_AMQP_ERROR_INVALID_FIELD,
-                   "Link property 'paired' is not set to boolean value 'true'", []).
+-spec error_property_paired_not_set() -> no_return().
+error_property_paired_not_set() ->
+    link_error(?V_1_0_AMQP_ERROR_INVALID_FIELD,
+               "Link property 'paired' is not set to boolean value 'true'", []).
 
 -spec exit_not_implemented(io:format()) -> no_return().
 exit_not_implemented(Format) ->
     exit_not_implemented(Format, []).
 
 -spec exit_not_implemented(io:format(), [term()]) -> no_return().
 exit_not_implemented(Format, Args) ->
-    protocol_error(?V_1_0_AMQP_ERROR_NOT_IMPLEMENTED, Format, Args).
+    link_error(?V_1_0_AMQP_ERROR_NOT_IMPLEMENTED, Format, Args).
 
 -spec error_not_found(rabbit_types:r(exchange | queue)) -> #'v1_0.error'{}.
 error_not_found(Resource) ->

deps/rabbit/test/amqp_client_SUITE.erl

@@ -994,46 +994,46 @@ durable_field(Config, QueueType, QName)
 invalid_transfer_settled_flag(Config) ->
     OpnConf = connection_config(Config),
     {ok, Connection} = amqp10_client:open_connection(OpnConf),
-    {ok, Session1} = amqp10_client:begin_session(Connection),
-    {ok, Session2} = amqp10_client:begin_session(Connection),
+    {ok, Session} = amqp10_client:begin_session(Connection),
     TargetAddr = rabbitmq_amqp_address:exchange(<<"amq.fanout">>),
     {ok, SenderSettled} = amqp10_client:attach_sender_link_sync(
-                            Session1, <<"link 1">>, TargetAddr, settled),
+                            Session, <<"link 1">>, TargetAddr, settled),
     {ok, SenderUnsettled} = amqp10_client:attach_sender_link_sync(
-                              Session2, <<"link 2">>, TargetAddr, unsettled),
+                              Session, <<"link 2">>, TargetAddr, unsettled),
     ok = wait_for_credit(SenderSettled),
     ok = wait_for_credit(SenderUnsettled),
 
     ok = amqp10_client:send_msg(SenderSettled, amqp10_msg:new(<<"tag1">>, <<"m1">>, false)),
     receive
         {amqp10_event,
-         {session, Session1,
-          {ended,
+         {link, SenderSettled,
+          {detached,
            #'v1_0.error'{
-              condition = ?V_1_0_CONNECTION_ERROR_FRAMING_ERROR,
+              condition = ?V_1_0_AMQP_ERROR_INVALID_FIELD,
               description = {utf8, Description1}}}}} ->
             ?assertEqual(
                <<"sender settle mode is 'settled' but transfer settled flag is interpreted as being 'false'">>,
                Description1)
-    after 30000 -> flush(missing_ended),
+    after 9000 -> flush(missing_event),
                   ct:fail({missing_event, ?LINE})
     end,
 
     ok = amqp10_client:send_msg(SenderUnsettled, amqp10_msg:new(<<"tag2">>, <<"m2">>, true)),
     receive
         {amqp10_event,
-         {session, Session2,
-          {ended,
+         {link, SenderUnsettled,
+          {detached,
            #'v1_0.error'{
-              condition = ?V_1_0_CONNECTION_ERROR_FRAMING_ERROR,
+              condition = ?V_1_0_AMQP_ERROR_INVALID_FIELD,
               description = {utf8, Description2}}}}} ->
             ?assertEqual(
                <<"sender settle mode is 'unsettled' but transfer settled flag is interpreted as being 'true'">>,
                Description2)
-    after 30000 -> flush(missing_ended),
+    after 9000 -> flush(missing_event),
                   ct:fail({missing_event, ?LINE})
     end,
 
+    ok = end_session_sync(Session),
     ok = close_connection_sync(Connection).
 
 quorum_queue_rejects(Config) ->
@@ -4664,15 +4664,16 @@ user_id(Config) ->
     ok = amqp10_client:send_msg(Sender, Msg2),
     receive
         {amqp10_event,
-         {session, Session,
-          {ended,
+         {link, Sender,
+          {detached,
            #'v1_0.error'{
               condition = ?V_1_0_AMQP_ERROR_UNAUTHORIZED_ACCESS,
               description = {utf8, <<"user_id property set to 'fake user' but authenticated user was 'guest'">>}}}}} -> ok
-    after 30000 -> flush(missing_ended),
+    after 9000 -> flush(missing_detached),
                   ct:fail("did not receive expected error")
     end,
 
+    ok = end_session_sync(Session),
     ok = close_connection_sync(Connection).
 
 message_ttl(Config) ->
@@ -6777,8 +6778,7 @@ attach_to_down_quorum_queue(Config) ->
     %% Create quorum queue with single replica on node 2.
     {_, _, LinkPair2} = Init2 = init(2, Config),
     {ok, _} = rabbitmq_amqp_client:declare_queue(
-                LinkPair2,
-                QName,
+                LinkPair2, QName,
                 #{arguments => #{<<"x-queue-type">> => {utf8, <<"quorum">>},
                                  <<"x-quorum-initial-group-size">> => {ulong, 1}
                                 }}),
@@ -6787,16 +6787,14 @@ attach_to_down_quorum_queue(Config) ->
     %% Make quorum queue unavailable.
     ok = rabbit_ct_broker_helpers:stop_broker(Config, 2),
 
-    OpnConf = connection_config(0, Config),
-    {ok, Connection} = amqp10_client:open_connection(OpnConf),
-    {ok, Session0} = amqp10_client:begin_session_sync(Connection),
+    {_, Session, LinkPair0} = Init0 = init(0, Config),
     flush(attaching_receiver),
-    {ok, _Receiver} = amqp10_client:attach_receiver_link(
-                        Session0, <<"receiver">>, Address),
+    {ok, Receiver} = amqp10_client:attach_receiver_link(
+                       Session, <<"receiver">>, Address),
     receive
         {amqp10_event,
-         {session, Session0,
-          {ended,
+         {link, Receiver,
+          {detached,
            #'v1_0.error'{
               condition = ?V_1_0_AMQP_ERROR_INTERNAL_ERROR,
               description = {utf8, Desc}}}}} ->
@@ -6808,12 +6806,8 @@ attach_to_down_quorum_queue(Config) ->
     end,
 
     ok = rabbit_ct_broker_helpers:start_broker(Config, 2),
-
-    {ok, Session} = amqp10_client:begin_session_sync(Connection),
-    {ok, LinkPair} = rabbitmq_amqp_client:attach_management_link_pair_sync(
-                       Session, <<"my link pair">>),
-    {ok, _} = rabbitmq_amqp_client:delete_queue(LinkPair, QName),
-    ok = close({Connection, Session, LinkPair}).
+    {ok, _} = rabbitmq_amqp_client:delete_queue(LinkPair0, QName),
+    ok = close(Init0).
 
 %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
 %% internal