Add outgoing messages interception call

Author: LoisSotoLopez

deps/rabbit/src/rabbit_amqp_session.erl

@@ -2164,7 +2164,8 @@ handle_deliver(ConsumerTag, AckRequired,
                                          conn_name = ConnName,
                                          channel_num = ChannelNum,
                                          user = #user{username = Username},
-                                         trace_state = Trace}}) ->
+                                         trace_state = Trace,
+                                         msg_interceptor_ctx = MsgInterceptorCtx}}) ->
     Handle = ctag_to_handle(ConsumerTag),
     case OutgoingLinks0 of
         #{Handle := #outgoing_link{queue_type = QType,
@@ -2180,7 +2181,10 @@ handle_deliver(ConsumerTag, AckRequired,
                           message_format = ?UINT(?MESSAGE_FORMAT),
                           settled = SendSettled},
             Mc1 = mc:convert(mc_amqp, Mc0),
-            Mc = mc:set_annotation(redelivered, Redelivered, Mc1),
+            Mc2 = rabbit_message_interceptor:intercept(Mc1,
+                                                       MsgInterceptorCtx,
+                                                       outgoing_message_interceptors),
+            Mc = mc:set_annotation(redelivered, Redelivered, Mc2),
             Sections = mc:protocol_state(Mc),
             validate_message_size(Sections, MaxMessageSize),
             Frames = transfer_frames(Transfer, Sections, MaxFrameSize),

deps/rabbit/src/rabbit_channel.erl

@@ -1255,7 +1255,8 @@ handle_method(#'basic.get'{queue = QueueNameBin, no_ack = NoAck},
                                          conn_pid = ConnPid,
                                          user = User,
                                          virtual_host = VHostPath,
-                                         authz_context = AuthzContext
+                                         authz_context = AuthzContext,
+                                         msg_interceptor_ctx = MsgInterceptorCtx
                                         },
                              limiter = Limiter,
                              next_tag   = DeliveryTag,
@@ -1270,8 +1271,11 @@ handle_method(#'basic.get'{queue = QueueNameBin, no_ack = NoAck},
                      Q, NoAck, rabbit_limiter:pid(Limiter),
                      DeliveryTag, QueueStates0)
            end) of
-        {ok, MessageCount, Msg, QueueStates} ->
+        {ok, MessageCount, Msg0, QueueStates} ->
             {ok, QueueType} = rabbit_queue_type:module(QueueName, QueueStates),
+            Msg = rabbit_message_interceptor:intercept(Msg0,
+                                                       MsgInterceptorCtx,
+                                                       outgoing_message_interceptors),
             handle_basic_get(WriterPid, DeliveryTag, NoAck, MessageCount, Msg,
                              QueueType, State#ch{queue_states = QueueStates});
         {empty, QueueStates} ->
@@ -2601,11 +2605,16 @@ handle_deliver(CTag, Ack, Msgs, State) when is_list(Msgs) ->
                 end, State, Msgs).
 
 handle_deliver0(ConsumerTag, AckRequired,
-                {QName, QPid, _MsgId, Redelivered, MsgCont0} = Msg,
+                {QName, QPid, _MsgId, Redelivered, MsgCont0} = Msg0,
                State = #ch{cfg = #conf{writer_pid = WriterPid,
-                                       writer_gc_threshold = GCThreshold},
+                                       writer_gc_threshold = GCThreshold,
+                                       msg_interceptor_ctx = MsgInterceptorCtx},
                            next_tag   = DeliveryTag,
                            queue_states = Qs}) ->
+
+    Msg = rabbit_message_interceptor:intercept(Msg0,
+                                               MsgInterceptorCtx,
+                                               outgoing_message_interceptors),
     Exchange = mc:exchange(MsgCont0),
     [RoutingKey | _] = mc:routing_keys(MsgCont0),
     MsgCont = mc:convert(mc_amqpl, MsgCont0),

deps/rabbitmq_mqtt/src/rabbit_mqtt_processor.erl

@@ -2073,7 +2073,10 @@ deliver_one_to_client({QNameOrType, QPid, QMsgId, _Redelivered, Mc} = Delivery,
                         true -> ?QOS_1;
                         false -> ?QOS_0
                     end,
-    McMqtt = mc:convert(mc_mqtt, Mc, mc_env()),
+    McMqtt0 = mc:convert(mc_mqtt, Mc, mc_env()),
+    McMqtt = rabbit_message_interceptor:intercept(McMqtt0,
+                                               build_msg_interceptor_ctx(State0),
+                                               outgoing_message_interceptors),
     MqttMsg = #mqtt_msg{qos = PublisherQos} = mc:protocol_state(McMqtt),
     QoS = effective_qos(PublisherQos, SubscriberQoS),
     {SettleOp, State1} = maybe_publish_to_client(MqttMsg, Delivery, QoS, State0),