Remove #outgoing_link.default_outcome

Author: ansd

deps/rabbitmq_amqp1_0/app.bzl

@@ -11,7 +11,6 @@ def all_beam_files(name = "all_beam_files"):
         srcs = [
             "src/Elixir.RabbitMQ.CLI.Ctl.Commands.ListAmqp10ConnectionsCommand.erl",
             "src/rabbit_amqp1_0.erl",
-            "src/rabbit_amqp1_0_link_util.erl",
             "src/rabbit_amqp1_0_reader.erl",
             "src/rabbit_amqp1_0_session.erl",
             "src/rabbit_amqp1_0_session_sup.erl",
@@ -43,7 +42,6 @@ def all_test_beam_files(name = "all_test_beam_files"):
         srcs = [
             "src/Elixir.RabbitMQ.CLI.Ctl.Commands.ListAmqp10ConnectionsCommand.erl",
             "src/rabbit_amqp1_0.erl",
-            "src/rabbit_amqp1_0_link_util.erl",
             "src/rabbit_amqp1_0_reader.erl",
             "src/rabbit_amqp1_0_session.erl",
             "src/rabbit_amqp1_0_session_sup.erl",
@@ -85,7 +83,6 @@ def all_srcs(name = "all_srcs"):
         srcs = [
             "src/Elixir.RabbitMQ.CLI.Ctl.Commands.ListAmqp10ConnectionsCommand.erl",
             "src/rabbit_amqp1_0.erl",
-            "src/rabbit_amqp1_0_link_util.erl",
             "src/rabbit_amqp1_0_reader.erl",
             "src/rabbit_amqp1_0_session.erl",
             "src/rabbit_amqp1_0_session_sup.erl",

deps/rabbitmq_amqp1_0/src/rabbit_amqp1_0_link_util.erl

@@ -19,6 +19,12 @@
 -define(INIT_TXFR_COUNT, 0).
 -define(DEFAULT_SEND_SETTLED, false).
 
+%% [3.4]
+-define(OUTCOMES, [?V_1_0_SYMBOL_ACCEPTED,
+                   ?V_1_0_SYMBOL_REJECTED,
+                   ?V_1_0_SYMBOL_RELEASED,
+                   ?V_1_0_SYMBOL_MODIFIED]).
+
 %% Just make these constant for the time being.
 -define(INCOMING_CREDIT, 65536).
 
@@ -41,8 +47,6 @@
 
 -import(rabbit_amqp1_0_util, [protocol_error/3,
                               serial_add/2, serial_diff/2, serial_compare/2]).
--import(rabbit_amqp1_0_link_util, [handle_to_ctag/1,
-                                   ctag_to_handle/1]).
 
 -record(incoming_link, {
           name,
@@ -58,9 +62,8 @@
 -record(outgoing_link, {
           queue :: undefined | rabbit_misc:resource_name(),
           delivery_count = 0,
-          %% TODO below 2 fields are not needed?
-          send_settled,
-          default_outcome}).
+          %% TODO below field is not needed?
+          send_settled}).
 
 -record(outgoing_unsettled, {
           %% The queue sent us this consumer scoped sequence number.
@@ -316,7 +319,7 @@ handle_control(#'v1_0.attach'{role = ?SEND_ROLE,
                                           routing_key = RoutingKey,
                                           delivery_count = InitTransfer,
                                           recv_settle_mode = RcvSettleMode},
-            {_DefaultOutcome, _Outcomes} = rabbit_amqp1_0_link_util:outcomes(Source),
+            _Outcomes = outcomes(Source),
             Confirm = case SndSettleMode of
                           ?V_1_0_SENDER_SETTLE_MODE_UNSETTLED -> true;
                           ?V_1_0_SENDER_SETTLE_MODE_SETTLED -> false;
@@ -359,13 +362,11 @@ handle_control(#'v1_0.attach'{role = ?RECV_ROLE,
                       queue_states = QStates0} = State0) ->
 
     ok = validate_attach(Attach),
-    {DefaultOutcome, Outcomes} = rabbit_amqp1_0_link_util:outcomes(Source),
     SndSettled = case SndSettleMode of
                      ?V_1_0_SENDER_SETTLE_MODE_SETTLED -> true;
                      ?V_1_0_SENDER_SETTLE_MODE_UNSETTLED -> false;
                      _ -> ?DEFAULT_SEND_SETTLED
                  end,
-    DOSym = amqp10_framing:symbol_for(DefaultOutcome),
     case ensure_source(Source, Vhost) of
         {ok, QNameBin} ->
             CTag = handle_to_ctag(Handle),
@@ -400,13 +401,16 @@ handle_control(#'v1_0.attach'{role = ?RECV_ROLE,
                                                                           false -> ?V_1_0_SENDER_SETTLE_MODE_UNSETTLED
                                                                       end,
                                                     rcv_settle_mode = RcvSettleMode,
-                                                    source = Source#'v1_0.source'{default_outcome = DefaultOutcome,
-                                                                                  outcomes = Outcomes},
+                                                    %% The queue process monitors our session process. When our session process terminates
+                                                    %% (abnormally) any messages checked out to our session process will be requeued.
+                                                    %% That's why the we only support RELEASED as the default outcome.
+                                                    source = Source#'v1_0.source'{
+                                                                      default_outcome = #'v1_0.released'{},
+                                                                      outcomes = outcomes(Source)},
                                                     role = ?SEND_ROLE},
                                    OutLink = #outgoing_link{delivery_count = ?INIT_TXFR_COUNT,
                                                             queue = QNameBin,
-                                                            send_settled = SndSettled,
-                                                            default_outcome = DOSym},
+                                                            send_settled = SndSettled},
                                    {ok, [AttachReply], OutLink, State1};
                                {error, Reason} ->
                                    protocol_error(
@@ -1394,7 +1398,7 @@ declare_queue(QNameBin, Vhost, TerminusDurability) ->
     rabbit_core_metrics:queue_declared(QName),
     Q0 = amqqueue:new(QName,
                       _Pid = none,
-                      rabbit_amqp1_0_link_util:durable(TerminusDurability),
+                      queue_is_durable(TerminusDurability),
                       _AutoDelete = false,
                       _QOwner = none,
                       _QArgs = [],
@@ -1411,6 +1415,47 @@ declare_queue(QNameBin, Vhost, TerminusDurability) ->
             protocol_error(?V_1_0_AMQP_ERROR_INTERNAL_ERROR, "Failed to declare ~s: ~p", [rabbit_misc:rs(QName), Other])
     end.
 
+outcomes(#'v1_0.source'{outcomes = undefined}) ->
+    {array, symbol, ?OUTCOMES};
+outcomes(#'v1_0.source'{outcomes = {array, symbol, Syms} = Outcomes}) ->
+    case lists:filter(fun(O) -> not lists:member(O, ?OUTCOMES) end, Syms) of
+        [] ->
+            Outcomes;
+        Unsupported ->
+            rabbit_amqp1_0_util:protocol_error(
+              ?V_1_0_AMQP_ERROR_NOT_IMPLEMENTED,
+              "Outcomes not supported: ~tp",
+              [Unsupported])
+    end;
+outcomes(#'v1_0.source'{outcomes = Unsupported}) ->
+    rabbit_amqp1_0_util:protocol_error(
+      ?V_1_0_AMQP_ERROR_NOT_IMPLEMENTED,
+      "Outcomes not supported: ~tp",
+      [Unsupported]);
+outcomes(_) ->
+    {array, symbol, ?OUTCOMES}.
+
+-spec handle_to_ctag({uint, non_neg_integer()}) ->
+    binary().
+handle_to_ctag({uint, H}) ->
+    <<"ctag-", H:32/integer>>.
+
+-spec ctag_to_handle(binary()) ->
+    {uint, non_neg_integer()}.
+ctag_to_handle(<<"ctag-", H:32/integer>>) ->
+    {uint, H}.
+
+queue_is_durable(?V_1_0_TERMINUS_DURABILITY_NONE) ->
+    false;
+queue_is_durable(?V_1_0_TERMINUS_DURABILITY_CONFIGURATION) ->
+    true;
+queue_is_durable(?V_1_0_TERMINUS_DURABILITY_UNSETTLED_STATE) ->
+    true;
+queue_is_durable(undefined) ->
+    %% <field name="durable" type="terminus-durability" default="none"/>
+    %% [3.5.3]
+    queue_is_durable(?V_1_0_TERMINUS_DURABILITY_NONE).
+
 %%% TODO move copied code to some common module
 %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
 %% BEGIN copy from rabbit_channel %%%

deps/rabbitmq_amqp1_0/src/rabbit_amqp1_0_session.erl

@@ -849,7 +849,6 @@ rabbit_common:
 rabbitmq_amqp1_0:
 - Elixir.RabbitMQ.CLI.Ctl.Commands.ListAmqp10ConnectionsCommand
 - rabbit_amqp1_0
-- rabbit_amqp1_0_link_util
 - rabbit_amqp1_0_reader
 - rabbit_amqp1_0_session
 - rabbit_amqp1_0_session_sup