Put proto value in interceptor ctx map

Author: LoisSotoLopez

deps/rabbit/src/rabbit_amqp_session.erl

@@ -476,7 +476,8 @@ init({ReaderPid, WriterPid, ChannelNum, MaxFrameSize, User, Vhost, ContainerId,
                            max_incoming_window = MaxIncomingWindow,
                            max_link_credit = MaxLinkCredit,
                            max_queue_credit = MaxQueueCredit,
-                           msg_interceptor_ctx = #{username => User#user.username,
+                           msg_interceptor_ctx = #{protocol => amqp,
+                                                   username => User#user.username,
                                                    vhost => Vhost,
                                                    conn_name => ConnName}
                           }}}.

deps/rabbit/src/rabbit_channel.erl

@@ -511,7 +511,8 @@ init([Channel, ReaderPid, WriterPid, ConnPid, ConnName, Protocol, User, VHost,
                             authz_context = OptionalVariables,
                             max_consumers = MaxConsumers,
                             writer_gc_threshold = GCThreshold,
-                            msg_interceptor_ctx = #{username => User#user.username,
+                            msg_interceptor_ctx = #{protocol => amqp091,
+                                                    username => User#user.username,
                                                     vhost => VHost,
                                                     conn_name => ConnName}
                            },

deps/rabbit/src/rabbit_message_interceptor.erl

@@ -3,7 +3,8 @@
 -export([intercept/3,
          set_msg_annotation/4]).
 
--type protocol() :: amqp091 | amqp | mqttv3 | mqttv5.
+-type protocol() :: amqp091 | amqp | mqtt310 | mqtt311 | mqtt50.
+
 -type msg_interceptor_ctx() :: #{protocol := protocol(),
                                  vhost := binary(),
                                  username := binary(),

deps/rabbitmq_mqtt/src/rabbit_mqtt_processor.erl

@@ -215,9 +215,10 @@ process_connect(
         %% To simplify logic, we decide at connection establishment time to stick
         %% with either binding args v1 or v2 for the lifetime of the connection.
         BindingArgsV2 = rabbit_feature_flags:is_enabled('rabbitmq_4.1.0'),
+        AtomProtoVer = proto_integer_to_atom(ProtoVer),
         S = #state{
                cfg = #cfg{socket = Socket,
-                          proto_ver = proto_integer_to_atom(ProtoVer),
+                          proto_ver = AtomProtoVer,
                           clean_start = CleanStart,
                           session_expiry_interval_secs = SessionExpiry,
                           ssl_login_name = SslLoginName,
@@ -239,7 +240,8 @@ process_connect(
                           max_packet_size_outbound = MaxPacketSize,
                           topic_alias_maximum_outbound = TopicAliasMaxOutbound,
                           binding_args_v2 = BindingArgsV2,
-                          msg_interceptor_ctx = #{user => Username,
+                          msg_interceptor_ctx = #{protocol => AtomProtoVer,
+                                                  username => Username,
                                                   vhost => VHost,
                                                   conn_name => ConnName,
                                                   client_id => ClientId}},