Simplify rabbit_channel

by removing extra AMQP 1.0 logic for settling unroutable messages with
released state.

This commit reverts the workaround introduced by PR 8015.

Author: ansd

deps/rabbit/src/rabbit_channel.erl

@@ -108,11 +108,7 @@
           consumer_timeout,
           authz_context,
           %% defines how ofter gc will be executed
-          writer_gc_threshold,
-          %% TODO delete
-          %% true with AMQP 1.0 to include the publishing sequence
-          %% in the return callback, false otherwise
-          extended_return_callback
+          writer_gc_threshold
          }).
 
 -record(pending_ack, {
@@ -513,7 +509,6 @@ init([Channel, ReaderPid, WriterPid, ConnPid, ConnName, Protocol, User, VHost,
     MaxMessageSize = get_max_message_size(),
     ConsumerTimeout = get_consumer_timeout(),
     OptionalVariables = extract_variable_map_from_amqp_params(AmqpParams),
-    UseExtendedReturnCallback = use_extended_return_callback(AmqpParams),
     {ok, GCThreshold} = application:get_env(rabbit, writer_gc_threshold),
     State = #ch{cfg = #conf{state = starting,
                             protocol = Protocol,
@@ -532,8 +527,7 @@ init([Channel, ReaderPid, WriterPid, ConnPid, ConnName, Protocol, User, VHost,
                             max_message_size = MaxMessageSize,
                             consumer_timeout = ConsumerTimeout,
                             authz_context = OptionalVariables,
-                            writer_gc_threshold = GCThreshold,
-                            extended_return_callback = UseExtendedReturnCallback
+                            writer_gc_threshold = GCThreshold
                            },
                 limiter = Limiter,
                 tx                      = none,
@@ -1043,17 +1037,6 @@ extract_variable_map_from_amqp_params([Value]) ->
 extract_variable_map_from_amqp_params(_) ->
     #{}.
 
-%%TODO delete
-%%
-%% Use tuple representation of amqp_params to avoid a dependency on amqp_client.
-%% Used for AMQP 1.0
-use_extended_return_callback({amqp_params_direct,_,_,_,_,
-                              {amqp_adapter_info,_,_,_,_,_,{'AMQP',"1.0"},_},
-                              _}) ->
-    true;
-use_extended_return_callback(_) ->
-    false.
-
 check_msg_size(Content, MaxMessageSize, GCThreshold) ->
     Size = rabbit_basic:maybe_gc_large_msg(Content, GCThreshold),
     case Size of
@@ -1272,15 +1255,15 @@ handle_method(#'basic.publish'{exchange    = ExchangeNameBin,
     check_user_id_header(Props, State),
     check_expiration_header(Props),
     DoConfirm = Tx =/= none orelse ConfirmEnabled,
-    {DeliveryOptions, SeqNum, State1} =
+    {DeliveryOptions, State1} =
         case DoConfirm of
             false ->
-                {maps_put_truthy(flow, Flow, #{}), undefined, State0};
+                {maps_put_truthy(flow, Flow, #{}), State0};
             true  ->
                 rabbit_global_counters:messages_received_confirm(amqp091, 1),
                 SeqNo = State0#ch.publish_seqno,
                 Opts = maps_put_truthy(flow, Flow, #{correlation => SeqNo}),
-                {Opts, SeqNo, State0#ch{publish_seqno = SeqNo + 1}}
+                {Opts, State0#ch{publish_seqno = SeqNo + 1}}
         end,
     % rabbit_feature_flags:is_enabled(message_containers),
     Message0 = mc_amqpl:message(ExchangeName,
@@ -1291,7 +1274,7 @@ handle_method(#'basic.publish'{exchange    = ExchangeNameBin,
     [rabbit_channel:deliver_reply(RK, Message) ||
      {virtual_reply_queue, RK} <- QNames],
     Queues = rabbit_amqqueue:lookup_many(QNames),
-    ok = process_routing_mandatory(Mandatory, Queues, SeqNum, Message, ExchangeName, State0),
+    ok = process_routing_mandatory(Mandatory, Queues, Message, ExchangeName, State0),
     rabbit_trace:tap_in(Message, QNames, ConnName, ChannelNum,
                         Username, TraceState),
     %% TODO: call delivery_to_queues with plain args
@@ -2191,32 +2174,23 @@ deliver_to_queues(XName,
 
 process_routing_mandatory(_Mandatory = true,
                           _RoutedToQs = [],
-                          MsgSeqNo,
                           Msg,
                           XName,
-                          State = #ch{cfg = #conf{extended_return_callback = ExtRetCallback}}) ->
+                          State) ->
     rabbit_global_counters:messages_unroutable_returned(amqp091, 1),
     ?INCR_STATS(exchange_stats, XName, 1, return_unroutable, State),
-    Content0 = mc:protocol_state(Msg),
-    Content = case ExtRetCallback of
-                  true ->
-                      %% providing the publishing sequence for AMQP 1.0
-                      {MsgSeqNo, Content0};
-                  false ->
-                      Content0
-              end,
+    Content = mc:protocol_state(Msg),
     [RoutingKey | _] = mc:get_annotation(routing_keys, Msg),
     ok = basic_return(Content, RoutingKey, XName#resource.name, State, no_route);
 process_routing_mandatory(_Mandatory = false,
                           _RoutedToQs = [],
-                          _MsgSeqNo,
                           _Msg,
                           XName,
                           State) ->
     rabbit_global_counters:messages_unroutable_dropped(amqp091, 1),
     ?INCR_STATS(exchange_stats, XName, 1, drop_unroutable, State),
     ok;
-process_routing_mandatory(_, _, _, _, _, _) ->
+process_routing_mandatory(_, _, _, _, _) ->
     ok.
 
 process_routing_confirm(undefined, _, _, State) ->

deps/rabbit_common/src/rabbit_writer.erl

@@ -105,10 +105,7 @@
 
 -spec send_command(pid(), rabbit_framing:amqp_method_record()) -> 'ok'.
 -spec send_command
-        (pid(), rabbit_framing:amqp_method_record(),
-         rabbit_types:content() |
-         {integer(), rabbit_types:content()} %% publishing sequence for AMQP 1.0 return callback
-        ) ->
+        (pid(), rabbit_framing:amqp_method_record(), rabbit_types:content()) ->
             'ok'.
 -spec send_command_sync(pid(), rabbit_framing:amqp_method_record()) -> 'ok'.
 -spec send_command_sync

deps/rabbitmq_amqp1_0/src/rabbit_amqp1_0_session.erl

@@ -839,24 +839,6 @@ handle_queue_event({queue_event, QRef, Evt},
             %     rabbit_misc:protocol_error(Type, Reason, ReasonArgs)
     end.
 
-% handle_info({#'basic.return'{}, {DTag, _Msg}}, State = #state{writer_pid = WriterPid,
-%                                                               session    = Session}) ->
-%     {Reply, Session1} = rabbit_amqp1_0_session:return(DTag, Session),
-%     case Reply of
-%         undefined ->
-%             ok;
-%         _ ->
-%             rabbit_amqp1_0_writer:send_command(
-%               WriterPid,
-%               rabbit_amqp1_0_session:flow_fields(Reply, Session)
-%              )
-%     end,
-%     {noreply, state(Session1, State)};
-
-% handle_info({#'basic.return'{}, _Msg}, State = #state{session = Session}) ->
-%     rabbit_log:warning("AMQP 1.0 message return without publishing sequence"),
-%     {noreply, state(Session, State)};
-
 handle_queue_actions(Actions, State0) ->
     {ReplyRev, State} =
     lists:foldl(
@@ -1100,7 +1082,6 @@ process_routing_confirm([], _SenderSettles = false, DeliveryId, _, U) ->
                                       first = {uint, DeliveryId},
                                       settled = true,
                                       state = #'v1_0.released'{}},
-
     {U, [Disposition]};
 process_routing_confirm([_|_] = Qs, _SenderSettles = false, DeliveryId, XName, U) ->
     QNames = rabbit_amqqueue:queue_names(Qs),