check topic alias

src/emqx_mqtt_caps.erl

@@ -43,7 +43,9 @@
                        {mqtt_wildcard_subscription, true}]).
 
 -define(PUBCAP_KEYS, [max_qos_allowed,
-                      mqtt_retain_available]).
+                      mqtt_retain_available,
+                      mqtt_topic_alias
+                     ]).
 -define(SUBCAP_KEYS, [max_qos_allowed,
                       max_topic_levels,
                       mqtt_shared_subscription,
@@ -60,6 +62,11 @@ do_check_pub(Props = #{qos := QoS}, [{max_qos_allowed, MaxQoS}|Caps]) ->
         true  -> {error, ?RC_QOS_NOT_SUPPORTED};
         false -> do_check_pub(Props, Caps)
     end;
+do_check_pub(Props = #{ topic_alias := TopicAlias}, [{max_topic_alias, MaxTopicAlias}| Caps]) ->
+    case TopicAlias =< MaxTopicAlias andalso TopicAlias > 0 of
+        false -> {error, ?RC_TOPIC_ALIAS_INVALID};
+        true -> do_check_pub(Props, Caps)
+    end;
 do_check_pub(#{retain := true}, [{mqtt_retain_available, false}|_Caps]) ->
     {error, ?RC_RETAIN_NOT_SUPPORTED};
 do_check_pub(Props, [{mqtt_retain_available, _}|Caps]) ->
@@ -136,4 +143,3 @@ with_env(Zone, Key, InitFun) ->
                      Caps;
         ZoneCaps  -> ZoneCaps
     end.
-

src/emqx_protocol.erl

@@ -33,36 +33,36 @@
 -export([shutdown/2]).
 
 -record(pstate, {
-          zone,
-          sendfun,
-          peername,
-          peercert,
-          proto_ver,
-          proto_name,
-          ackprops,
-          client_id,
-          is_assigned,
-          conn_pid,
-          conn_props,
-          ack_props,
-          username,
-          session,
-          clean_start,
-          topic_aliases,
-          packet_size,
-          will_topic,
-          will_msg,
-          keepalive,
-          mountpoint,
-          is_super,
-          is_bridge,
-          enable_ban,
-          enable_acl,
-          recv_stats,
-          send_stats,
-          connected,
-          connected_at
-         }).
+                 zone,
+                 sendfun,
+                 peername,
+                 peercert,
+                 proto_ver,
+                 proto_name,
+                 ackprops,
+                 client_id,
+                 is_assigned,
+                 conn_pid,
+                 conn_props,
+                 ack_props,
+                 username,
+                 session,
+                 clean_start,
+                 topic_aliases,
+                 packet_size,
+                 will_topic,
+                 will_msg,
+                 keepalive,
+                 mountpoint,
+                 is_super,
+                 is_bridge,
+                 enable_ban,
+                 enable_acl,
+                 recv_stats,
+                 send_stats,
+                 connected,
+                 connected_at
+                }).
 
 -type(state() :: #pstate{}).
 -export_type([state/0]).
@@ -631,9 +631,11 @@ check_publish(Packet, PState) ->
     run_check_steps([fun check_pub_caps/2,
                      fun check_pub_acl/2], Packet, PState).
 
-check_pub_caps(#mqtt_packet{header = #mqtt_packet_header{qos = QoS, retain = R}},
+check_pub_caps(#mqtt_packet{header = #mqtt_packet_header{qos = QoS, retain = Retain},
+                            variable = #mqtt_packet_publish{ properties = Properties}},
                #pstate{zone = Zone}) ->
-    emqx_mqtt_caps:check_pub(Zone, #{qos => QoS, retain => R}).
+    #{'Topic-Alias' := TopicAlias} = Properties,
+    emqx_mqtt_caps:check_pub(Zone, #{qos => QoS, retain => Retain, topic_alias => TopicAlias}).
 
 check_pub_acl(_Packet, #pstate{is_super = IsSuper, enable_acl = EnableAcl})
     when IsSuper orelse (not EnableAcl) ->