Remove behaviour & use context map

Author: LoisSotoLopez

deps/rabbit/Makefile

@@ -273,7 +273,7 @@ PARALLEL_CT_SET_3_B = cluster_upgrade list_consumers_sanity_check list_queues_on
 PARALLEL_CT_SET_3_C = cli_forget_cluster_node feature_flags_v2 mc_unit message_containers_deaths_v2 message_size_limit metadata_store_migration
 PARALLEL_CT_SET_3_D = metadata_store_phase1 metrics mirrored_supervisor peer_discovery_classic_config proxy_protocol runtime_parameters unit_stats_and_metrics unit_supervisor2 unit_vm_memory_monitor
 
-PARALLEL_CT_SET_4_A = clustering_events rabbit_local_random_exchange rabbit_incoming_message_interceptor rabbitmq_4_0_deprecations unit_pg_local unit_plugin_directories unit_plugin_versioning unit_policy_validators unit_priority_queue
+PARALLEL_CT_SET_4_A = clustering_events rabbit_local_random_exchange rabbit_message_interceptor rabbitmq_4_0_deprecations unit_pg_local unit_plugin_directories unit_plugin_versioning unit_policy_validators unit_priority_queue
 PARALLEL_CT_SET_4_B = per_user_connection_tracking per_vhost_connection_limit rabbit_fifo_dlx_integration rabbit_fifo_int
 PARALLEL_CT_SET_4_C = msg_size_metrics unit_msg_size_metrics per_vhost_msg_store per_vhost_queue_limit priority_queue upgrade_preparation vhost
 PARALLEL_CT_SET_4_D = per_user_connection_channel_tracking product_info publisher_confirms_parallel queue_type rabbitmq_queues_cli_integration rabbitmqctl_integration rabbitmqctl_shutdown routing rabbit_amqqueue

deps/rabbit/ct.test.spec

@@ -115,7 +115,7 @@
 ,   rabbit_fifo_prop_SUITE
 ,   rabbit_fifo_v0_SUITE
 ,   rabbit_local_random_exchange_SUITE
-,   rabbit_incoming_message_interceptor_SUITE
+,   rabbit_message_interceptor_SUITE
 ,   rabbit_stream_coordinator_SUITE
 ,   rabbit_stream_sac_coordinator_SUITE
 ,   rabbitmq_4_0_deprecations_SUITE

deps/rabbit/priv/schema/rabbit.schema

@@ -2658,14 +2658,17 @@ end}.
 {mapping, "message_interceptors.incoming.$interceptor.overwrite", "rabbit.incoming_message_interceptors", [
     {datatype, {enum, [true, false]}}]}.
 
-% Pseudo-key to include the interceptor in the list of interceptors. 
+% Pseudo-key to include the interceptor in the list of interceptors.
 %  - If any other configuration is provided for the interceptor this
-%  configuration is not required. 
+%  configuration is not required.
 %  - If no other configuration is provided, this one is required so that the
-%  interceptor gets invoked. 
+%  interceptor gets invoked.
 {mapping, "message_interceptors.incoming.$interceptor.enabled", "rabbit.incoming_message_interceptors", [
     {datatype, {enum, [true]}}]}.
 
+{mapping, "message_interceptors.outgoing.$interceptor.enabled", "rabbit.outgoing_message_interceptors", [
+    {datatype, {enum, [true]}}]}.
+
 {mapping, "message_interceptors.incoming.set_header_timestamp.overwrite", "rabbit.incoming_message_interceptors", [
     {datatype, {enum, [true, false]}}]}.
 
@@ -2674,19 +2677,21 @@ end}.
 
 {translation, "rabbit.incoming_message_interceptors",
     fun(Conf) ->
-         case cuttlefish_variable:filter_by_prefix("message_interceptors.incoming", Conf) of
-             [] ->
-                 cuttlefish:unset();
-             L ->
-                InterceptorsConfig = [ 
+        case cuttlefish_variable:filter_by_prefix("message_interceptors.incoming", Conf) of
+            [] ->
+                cuttlefish:unset();
+            L ->
+                InterceptorsConfig = [
                     {Module0, Config, Value}
                     || {["message_interceptors", "incoming", Module0, Config], Value} <- L
                 ],
                 {Result, Order0} = lists:foldl(
-                    fun({Interceptor0, Key0, Value}, {Acc, Order}) -> 
+                    fun({Interceptor0, Key0, Value}, {Acc, Order}) ->
                         Interceptor = list_to_atom(Interceptor0),
                         Key = list_to_atom(Key0),
-                        MapPutFun = fun(Key, Value) -> fun(Old) -> maps:put(Key, Value, Old) end end,
+                        MapPutFun = fun(Old) -> maps:put(Key, Value, Old) end,
+                        % This Interceptor -> Module alias exists for
+                        % compatibility reasons
                         Module = case Interceptor  of
                             set_header_timestamp ->
                                 rabbit_header_timestamp_interceptor;
@@ -2695,10 +2700,10 @@ end}.
                             _ ->
                                 Interceptor
                         end,
-                        NewAcc = 
+                        NewAcc =
                             maps:update_with(
-                                Module, 
-                                MapPutFun(Key, Value), 
+                                Module,
+                                MapPutFun,
                                 #{Key => Value},
                                 Acc),
                         {NewAcc, [Module| Order]}
@@ -2708,7 +2713,39 @@ end}.
                 ),
                 Order = lists:uniq(Order0),
                 [{O, maps:without([enabled], maps:get(O, Result))} || O <- Order]
-         end
+        end
+    end
+}.
+
+{translation, "rabbit.outgoing_message_interceptors",
+    fun(Conf) ->
+        case cuttlefish_variable:filter_by_prefix("message_interceptors.outgoing", Conf) of
+            [] ->
+                cuttlefish:unset();
+            L ->
+                InterceptorsConfig = [
+                    {Module0, Config, Value}
+                    || {["message_interceptors", "outgoing", Module0, Config], Value} <- L
+                ],
+                {Result, Order0} = lists:foldl(
+                    fun({Interceptor0, Key0, Value}, {Acc, Order}) ->
+                        Module = list_to_atom(Interceptor0),
+                        Key = list_to_atom(Key0),
+                        MapPutFun = fun(Old) -> maps:put(Key, Value, Old) end,
+                        NewAcc =
+                            maps:update_with(
+                                Module,
+                                MapPutFun,
+                                #{Key => Value},
+                                Acc),
+                        {NewAcc, [Module| Order]}
+                    end,
+                    {#{}, []},
+                    InterceptorsConfig
+                ),
+                Order = lists:uniq(Order0),
+                [{O, maps:without([enabled], maps:get(O, Result))} || O <- Order]
+        end
     end
 }.
 

deps/rabbit/src/rabbit_amqp_session.erl

@@ -10,7 +10,6 @@
 -compile({inline, [maps_update_with/4]}).
 
 -behaviour(gen_server).
--behaviour(rabbit_protocol_accessor).
 
 -include_lib("kernel/include/logger.hrl").
 -include_lib("rabbit_common/include/rabbit.hrl").
@@ -106,9 +105,6 @@
          handle_info/2,
          format_status/1]).
 
-% `rabbit_protocol_accessor` behaviour callbacks
--export([get_property/2]).
-
 -import(rabbit_amqp_util,
         [protocol_error/3]).
 -import(serial_number,
@@ -287,7 +283,8 @@
           max_handle :: link_handle(),
           max_incoming_window :: pos_integer(),
           max_link_credit :: pos_integer(),
-          max_queue_credit :: pos_integer()
+          max_queue_credit :: pos_integer(),
+          msg_interceptor_ctx :: map()
          }).
 
 -record(state, {
@@ -478,7 +475,10 @@ init({ReaderPid, WriterPid, ChannelNum, MaxFrameSize, User, Vhost, ContainerId,
                            max_handle = EffectiveHandleMax,
                            max_incoming_window = MaxIncomingWindow,
                            max_link_credit = MaxLinkCredit,
-                           max_queue_credit = MaxQueueCredit
+                           max_queue_credit = MaxQueueCredit,
+                           msg_interceptor_ctx = #{user => User,
+                                                   vhost => Vhost,
+                                                   conn_name => ConnName}
                           }}}.
 
 terminate(_Reason, #state{incoming_links = IncomingLinks,
@@ -2415,7 +2415,8 @@ incoming_link_transfer(
                              trace_state = Trace,
                              conn_name = ConnName,
                              channel_num = ChannelNum,
-                             max_link_credit = MaxLinkCredit}}) ->
+                             max_link_credit = MaxLinkCredit,
+                             msg_interceptor_ctx = MsgInterceptorCtx}}) ->
 
     {PayloadBin, DeliveryId, Settled} =
     case MultiTransfer of
@@ -2440,7 +2441,9 @@ incoming_link_transfer(
     Mc0 = mc:init(mc_amqp, PayloadBin, #{}),
     case lookup_target(LinkExchange, LinkRKey, Mc0, Vhost, User, PermCache0) of
         {ok, X, RoutingKeys, Mc1, PermCache} ->
-            Mc2 = rabbit_incoming_message_interceptor:intercept(Mc1, ?MODULE, State0),
+            Mc2 = rabbit_message_interceptor:intercept(Mc1,
+                                                       MsgInterceptorCtx,
+                                                       incoming),
             check_user_id(Mc2, User),
             TopicPermCache = check_write_permitted_on_topics(
                                X, User, RoutingKeys, TopicPermCache0),
@@ -3878,15 +3881,6 @@ format_status(
               topic_permission_cache => TopicPermissionCache},
     maps:update(state, State, Status).
 
-get_property(user, #state{cfg = #cfg{user = User}}) ->
-    User; 
-get_property(vhost, #state{cfg = #cfg{vhost = VHost}}) ->
-    VHost; 
-get_property(connection_name, #state{cfg = #cfg{conn_name = ConnectionName}}) ->
-    ConnectionName;
-get_property(_, _) ->
-    undefined.
-
 -spec info(pid()) ->
     {ok, rabbit_types:infos()} | {error, term()}.
 info(Pid) ->

deps/rabbit/src/rabbit_channel.erl

@@ -7,8 +7,6 @@
 
 -module(rabbit_channel).
 
--behaviour(rabbit_protocol_accessor).
-
 %% rabbit_channel processes represent an AMQP 0-9-1 channels.
 %%
 %% Connections parse protocol frames coming from clients and
@@ -62,9 +60,6 @@
          prioritise_call/4, prioritise_cast/3, prioritise_info/3,
          format_message_queue/2]).
 
-% `rabbit_protocol_accessor` behaviour callbacks
--export([get_property/2]).
-
 %% Internal
 -export([list_local/0, emit_info_local/3, deliver_reply_local/3]).
 -export([get_vhost/1, get_user/1]).
@@ -115,7 +110,8 @@
           authz_context,
           max_consumers,  % taken from rabbit.consumer_max_per_channel
           %% defines how ofter gc will be executed
-          writer_gc_threshold
+          writer_gc_threshold,
+          msg_interceptor_ctx
          }).
 
 -record(pending_ack, {
@@ -514,7 +510,10 @@ init([Channel, ReaderPid, WriterPid, ConnPid, ConnName, Protocol, User, VHost,
                             consumer_timeout = ConsumerTimeout,
                             authz_context = OptionalVariables,
                             max_consumers = MaxConsumers,
-                            writer_gc_threshold = GCThreshold
+                            writer_gc_threshold = GCThreshold,
+                            msg_interceptor_ctx = #{user => User,
+                                                   vhost => VHost,
+                                                   conn_name => ConnName}
                            },
                 limiter = Limiter,
                 tx                      = none,
@@ -819,15 +818,6 @@ get_consumer_timeout() ->
             undefined
     end.
 
-get_property(user, #ch{cfg = #conf{user = User}}) ->
-    User; 
-get_property(vhost, #ch{cfg = #conf{virtual_host = VHost}}) ->
-    VHost; 
-get_property(connection_name, #ch{cfg = #conf{conn_name = ConnectionName}}) ->
-    ConnectionName;
-get_property(_, _) ->
-    undefined.
-
 %%---------------------------------------------------------------------------
 
 reply(Reply, NewState) -> {reply, Reply, next_state(NewState), hibernate}.
@@ -1182,7 +1172,8 @@ handle_method(#'basic.publish'{exchange    = ExchangeNameBin,
                                                 user = #user{username = Username} = User,
                                                 trace_state = TraceState,
                                                 authz_context = AuthzContext,
-                                                writer_gc_threshold = GCThreshold
+                                                writer_gc_threshold = GCThreshold,
+                                                msg_interceptor_ctx = MsgInterceptorCtx
                                                },
                                    tx               = Tx,
                                    confirm_enabled  = ConfirmEnabled,
@@ -1221,7 +1212,9 @@ handle_method(#'basic.publish'{exchange    = ExchangeNameBin,
             rabbit_misc:precondition_failed("invalid message: ~tp", [Reason]);
         {ok, Message0} ->
             check_write_permitted_on_topics(Exchange, User, Message0, AuthzContext),
-            Message = rabbit_incoming_message_interceptor:intercept(Message0, ?MODULE, State),
+            Message = rabbit_message_interceptor:intercept(Message0,
+                                                           MsgInterceptorCtx,
+                                                           incoming),
             check_user_id_header(Message, User),
             QNames = rabbit_exchange:route(Exchange, Message, #{return_binding_keys => true}),
             [deliver_reply(RK, Message) || {virtual_reply_queue, RK} <- QNames],

deps/rabbit/src/rabbit_header_routing_node_interceptor.erl

@@ -1,16 +1,16 @@
 -module(rabbit_header_routing_node_interceptor).
--behaviour(rabbit_incoming_message_interceptor).
+-behaviour(rabbit_message_interceptor).
 
 -define(HEADER_ROUTING_NODE, <<"x-routed-by">>).
 
 -export([
-    intercept/4
+    intercept/3
 ]).
 
-intercept(Msg, _ProtoMod, _ProtoState, Config) ->
+intercept(Msg, _MsgInterceptorCtx, Config) ->
     Node = atom_to_binary(node()),
     Overwrite = maps:get(overwrite, Config, false),
-    rabbit_incoming_message_interceptor:set_msg_annotation(
+    rabbit_message_interceptor:set_msg_annotation(
         Msg,
         ?HEADER_ROUTING_NODE,
         Node,

deps/rabbit/src/rabbit_header_timestamp_interceptor.erl

@@ -1,18 +1,18 @@
 -module(rabbit_header_timestamp_interceptor).
--behaviour(rabbit_incoming_message_interceptor).
+-behaviour(rabbit_message_interceptor).
 
 -include("mc.hrl").
 
 -define(HEADER_TIMESTAMP, <<"timestamp_in_ms">>).
 
 -export([
-    intercept/4
+    intercept/3
 ]).
 
-intercept(Msg0, _ProtoMod, _ProtoState, Config) ->
+intercept(Msg0, _MsgInterceptorCtx, Config) ->
     Ts = mc:get_annotation(?ANN_RECEIVED_AT_TIMESTAMP, Msg0),
     Overwrite = maps:get(overwrite, Config, false),
-    Msg = rabbit_incoming_message_interceptor:set_msg_annotation(
+    Msg = rabbit_message_interceptor:set_msg_annotation(
         Msg0,
         ?HEADER_TIMESTAMP,
         Ts,