Improve topic alias maximum and receive maximum

src/emqx_inflight.erl

@@ -14,7 +14,7 @@
 
 -module(emqx_inflight).
 
--export([new/1, contain/2, lookup/2, insert/3, update/3, delete/2, values/1,
+-export([new/1, contain/2, lookup/2, insert/3, update/3, update_size/2, delete/2, values/1,
          to_list/1, size/1, max_size/1, is_full/1, is_empty/1, window/1]).
 
 -type(max_size() :: pos_integer()).
@@ -46,6 +46,10 @@ delete(Key, {?MODULE, MaxSize, Tree}) ->
 update(Key, Val, {?MODULE, MaxSize, Tree}) ->
     {?MODULE, MaxSize, gb_trees:update(Key, Val, Tree)}.
 
+-spec(update_size(integer(), inflight()) -> inflight()).
+update_size(MaxSize, {?MODULE, _OldMaxSize, Tree}) ->
+    {?MODULE, MaxSize, Tree}.
+
 -spec(is_full(inflight()) -> boolean()).
 is_full({?MODULE, 0, _Tree}) ->
     false;

src/emqx_mqtt_caps.erl

@@ -44,7 +44,7 @@
 
 -define(PUBCAP_KEYS, [max_qos_allowed,
                       mqtt_retain_available,
-                      mqtt_topic_alias
+                      max_topic_alias
                      ]).
 -define(SUBCAP_KEYS, [max_qos_allowed,
                       max_topic_levels,

src/emqx_packet.erl

@@ -61,6 +61,9 @@ validate(?PUBLISH_PACKET(_QoS, Topic, _, Properties, _)) ->
     ((not emqx_topic:wildcard(Topic)) orelse error(topic_name_invalid))
         andalso validate_properties(?PUBLISH, Properties);
 
+validate(?CONNECT_PACKET(#mqtt_packet_connect{properties = #{'Receive-Maximum' := 0}})) ->
+    error(protocol_error);
+
 validate(_Packet) ->
     true.
 

src/emqx_protocol.erl

@@ -208,11 +208,8 @@ received(Packet = ?PACKET(Type), PState) ->
         true ->
             {Packet1, PState1} = preprocess_properties(Packet, PState),
             process_packet(Packet1, inc_stats(recv, Type, PState1));
-        {'EXIT', {topic_filters_invalid, _Stacktrace}} ->
-            deliver({disconnect, ?RC_PROTOCOL_ERROR}, PState),
-            {error, topic_filters_invalid, PState};
         {'EXIT', {Reason, _Stacktrace}} ->
-            deliver({disconnect, ?RC_MALFORMED_PACKET}, PState),
+            deliver({disconnect, rc(Reason)}, PState),
             {error, Reason, PState}
     end.
 
@@ -415,12 +412,14 @@ process_packet(?DISCONNECT_PACKET(?RC_SUCCESS, #{'Session-Expiry-Interval' := In
     case Interval =/= 0 andalso OldInterval =:= 0 of
         true -> 
             deliver({disconnect, ?RC_PROTOCOL_ERROR}, PState),
-            {error, protocol_error, PState};
+            {error, protocol_error, PState#pstate{will_msg = undefined}};
         false -> 
             emqx_session:update_expiry_interval(SPid, Interval),
             %% Clean willmsg
             {stop, normal, PState#pstate{will_msg = undefined}}
     end;
+process_packet(?DISCONNECT_PACKET(?RC_SUCCESS), PState) ->
+    {stop, normal, PState#pstate{will_msg = undefined}};
 process_packet(?DISCONNECT_PACKET(_), PState) ->
     {stop, normal, PState}.
 
@@ -577,13 +576,11 @@ maybe_assign_client_id(PState = #pstate{client_id = <<>>, ackprops = AckProps})
 maybe_assign_client_id(PState) ->
     PState.
 
-try_open_session(#pstate{zone        = Zone,
-                         proto_ver   = ProtoVer,
-                         client_id   = ClientId,
-                         conn_pid    = ConnPid,
-                         conn_props  = ConnProps,
-                         username    = Username,
-                         clean_start = CleanStart}) ->
+try_open_session(PState = #pstate{zone        = Zone,
+                                  client_id   = ClientId,
+                                  conn_pid    = ConnPid,
+                                  username    = Username,
+                                  clean_start = CleanStart}) ->
 
     SessAttrs = #{
         zone        => Zone,
@@ -593,22 +590,42 @@ try_open_session(#pstate{zone        = Zone,
         clean_start => CleanStart
     },
 
-    case emqx_sm:open_session(maps:put(expiry_interval, if 
-                                ProtoVer =:= ?MQTT_PROTO_V5 ->
-                                    maps:get('Session-Expiry-Interval', ConnProps, 0);
-                                true ->
-                                    case CleanStart of
-                                        true ->
-                                            0;
-                                        false ->
-                                            emqx_zone:get_env(Zone, session_expiry_interval, 16#ffffffff)
-                                    end  
-                            end, SessAttrs)) of
+    SessAttrs1 = lists:foldl(fun set_session_attrs/2, SessAttrs, [{max_inflight, PState}, {expiry_interval, PState}, {topic_alias_maximum, PState}]),
+    case emqx_sm:open_session(SessAttrs1) of
         {ok, SPid} ->
             {ok, SPid, false};
         Other -> Other
     end.
 
+set_session_attrs({max_inflight, #pstate{zone = Zone, proto_ver = ProtoVer, conn_props = ConnProps}}, SessAttrs) ->
+    maps:put(max_inflight, if
+                               ProtoVer =:= ?MQTT_PROTO_V5 ->
+                                   maps:get('Receive-Maximum', ConnProps, 65535);
+                               true -> 
+                                   emqx_zone:get_env(Zone, max_inflight, 65535)
+                           end, SessAttrs);
+set_session_attrs({expiry_interval, #pstate{zone = Zone, proto_ver = ProtoVer, conn_props = ConnProps, clean_start = CleanStart}}, SessAttrs) ->
+    maps:put(expiry_interval, if
+                               ProtoVer =:= ?MQTT_PROTO_V5 ->
+                                   maps:get('Session-Expiry-Interval', ConnProps, 0);
+                               true -> 
+                                   case CleanStart of
+                                       true -> 0;
+                                       false -> 
+                                           emqx_zone:get_env(Zone, session_expiry_interval, 16#ffffffff)
+                                   end
+                           end, SessAttrs);
+set_session_attrs({topic_alias_maximum, #pstate{zone = Zone, proto_ver = ProtoVer, conn_props = ConnProps}}, SessAttrs) ->
+    maps:put(topic_alias_maximum, if
+                                    ProtoVer =:= ?MQTT_PROTO_V5 ->
+                                        maps:get('Topic-Alias-Maximum', ConnProps, 0);
+                                    true -> 
+                                        emqx_zone:get_env(Zone, max_topic_alias, 0)
+                                  end, SessAttrs);
+set_session_attrs({_, #pstate{}}, SessAttrs) ->
+    SessAttrs.
+
+
 authenticate(Credentials, Password) ->
     case emqx_access_control:authenticate(Credentials, Password) of
         ok -> {ok, false};
@@ -782,6 +799,14 @@ start_keepalive(Secs, #pstate{zone = Zone}) when Secs > 0 ->
     Backoff = emqx_zone:get_env(Zone, keepalive_backoff, 0.75),
     self() ! {keepalive, start, round(Secs * Backoff)}.
 
+rc(Reason) ->
+    case Reason of
+        protocol_error -> ?RC_PROTOCOL_ERROR;
+        topic_filters_invalid -> ?RC_TOPIC_FILTER_INVALID;
+        topic_name_invalid -> ?RC_TOPIC_NAME_INVALID;
+        _ -> ?RC_MALFORMED_PACKET
+    end.
+
 %%-----------------------------------------------------------------------------
 %% Parse topic filters
 %%-----------------------------------------------------------------------------

src/emqx_session.erl

@@ -47,7 +47,7 @@
 -export([info/1, attrs/1]).
 -export([stats/1]).
 -export([resume/2, discard/2]).
--export([update_expiry_interval/2]).
+-export([update_expiry_interval/2, update_misc/2]).
 -export([subscribe/2, subscribe/4]).
 -export([publish/3]).
 -export([puback/2, puback/3]).
@@ -145,7 +145,9 @@
           enqueue_stats = 0,
 
           %% Created at
-          created_at :: erlang:timestamp()
+          created_at :: erlang:timestamp(),
+
+          topic_alias_maximum :: pos_integer()
          }).
 
 -type(spid() :: pid()).
@@ -318,6 +320,9 @@ discard(SPid, ByPid) ->
 update_expiry_interval(SPid, Interval) ->
     gen_server:cast(SPid, {expiry_interval, Interval * 1000}).
 
+update_misc(SPid, Misc) ->
+    gen_server:cast(SPid, {update_misc, Misc}).
+
 -spec(close(spid()) -> ok).
 close(SPid) ->
     gen_server:call(SPid, close, infinity).
@@ -326,36 +331,38 @@ close(SPid) ->
 %% gen_server callbacks
 %%------------------------------------------------------------------------------
 
-init([Parent, #{zone            := Zone,
-                client_id       := ClientId,
-                username        := Username,
-                conn_pid        := ConnPid,
-                clean_start     := CleanStart,
-                expiry_interval := ExpiryInterval}]) ->
+init([Parent, #{zone                := Zone,
+                client_id           := ClientId,
+                username            := Username,
+                conn_pid            := ConnPid,
+                clean_start         := CleanStart,
+                expiry_interval     := ExpiryInterval,
+                max_inflight        := MaxInflight,
+                topic_alias_maximum := TopicAliasMaximum}]) ->
     process_flag(trap_exit, true),
     true = link(ConnPid),
-    MaxInflight = get_env(Zone, max_inflight),
     IdleTimout = get_env(Zone, idle_timeout, 30000),
-    State = #state{idle_timeout      = IdleTimout,
-                   clean_start       = CleanStart,
-                   binding           = binding(ConnPid),
-                   client_id         = ClientId,
-                   username          = Username,
-                   conn_pid          = ConnPid,
-                   subscriptions     = #{},
-                   max_subscriptions = get_env(Zone, max_subscriptions, 0),
-                   upgrade_qos       = get_env(Zone, upgrade_qos, false),
-                   inflight          = emqx_inflight:new(MaxInflight),
-                   mqueue            = init_mqueue(Zone),
-                   retry_interval    = get_env(Zone, retry_interval, 0),
-                   awaiting_rel      = #{},
-                   await_rel_timeout = get_env(Zone, await_rel_timeout),
-                   max_awaiting_rel  = get_env(Zone, max_awaiting_rel),
-                   expiry_interval   = ExpiryInterval,
-                   enable_stats      = get_env(Zone, enable_stats, true),
-                   deliver_stats     = 0,
-                   enqueue_stats     = 0,
-                   created_at        = os:timestamp()
+    State = #state{idle_timeout        = IdleTimout,
+                   clean_start         = CleanStart,
+                   binding             = binding(ConnPid),
+                   client_id           = ClientId,
+                   username            = Username,
+                   conn_pid            = ConnPid,
+                   subscriptions       = #{},
+                   max_subscriptions   = get_env(Zone, max_subscriptions, 0),
+                   upgrade_qos         = get_env(Zone, upgrade_qos, false),
+                   inflight            = emqx_inflight:new(MaxInflight),
+                   mqueue              = init_mqueue(Zone),
+                   retry_interval      = get_env(Zone, retry_interval, 0),
+                   awaiting_rel        = #{},
+                   await_rel_timeout   = get_env(Zone, await_rel_timeout),
+                   max_awaiting_rel    = get_env(Zone, max_awaiting_rel),
+                   expiry_interval     = ExpiryInterval,
+                   enable_stats        = get_env(Zone, enable_stats, true),
+                   deliver_stats       = 0,
+                   enqueue_stats       = 0,
+                   created_at          = os:timestamp(),
+                   topic_alias_maximum = TopicAliasMaximum
                   },
     emqx_sm:register_session(ClientId, attrs(State)),
     emqx_sm:set_session_stats(ClientId, stats(State)),
@@ -543,6 +550,10 @@ handle_cast({resume, ConnPid}, State = #state{client_id       = ClientId,
 handle_cast({expiry_interval, Interval}, State) ->
     {noreply, State#state{expiry_interval = Interval}};
 
+handle_cast({update_misc, #{max_inflight := MaxInflight, topic_alias_maximum := TopicAliasMaximum}}, State) ->
+    {noreply, State#state{inflight            = emqx_inflight:update_size(MaxInflight, State#state.inflight),
+                          topic_alias_maximum = TopicAliasMaximum}};
+
 handle_cast(Msg, State) ->
     emqx_logger:error("[Session] unexpected cast: ~p", [Msg]),
     {noreply, State}.
@@ -554,15 +565,22 @@ handle_info({dispatch, Topic, Msgs}, State) when is_list(Msgs) ->
                           end, State, Msgs)};
 
 %% Dispatch message
-handle_info({dispatch, Topic, Msg}, State = #state{subscriptions = SubMap}) when is_record(Msg, message) ->
-    noreply(case maps:find(Topic, SubMap) of
-                {ok, #{nl := Nl, qos := QoS, rap := Rap, subid := SubId}} ->
-                    run_dispatch_steps([{nl, Nl}, {qos, QoS}, {rap, Rap}, {subid, SubId}], Msg, State);
-                {ok, #{nl := Nl, qos := QoS, rap := Rap}} ->
-                    run_dispatch_steps([{nl, Nl}, {qos, QoS}, {rap, Rap}], Msg, State);
-                error ->
-                    dispatch(emqx_message:unset_flag(dup, Msg), State)
-            end);
+handle_info({dispatch, Topic, Msg = #message{headers = Headers}}, 
+            State = #state{subscriptions = SubMap, topic_alias_maximum = TopicAliasMaximum}) when is_record(Msg, message) ->
+    TopicAlias = maps:get('Topic-Alias', Headers, undefined),
+    if 
+        TopicAlias =:= undefined orelse TopicAlias =< TopicAliasMaximum ->
+            noreply(case maps:find(Topic, SubMap) of
+                        {ok, #{nl := Nl, qos := QoS, rap := Rap, subid := SubId}} ->
+                            run_dispatch_steps([{nl, Nl}, {qos, QoS}, {rap, Rap}, {subid, SubId}], Msg, State);
+                        {ok, #{nl := Nl, qos := QoS, rap := Rap}} ->
+                            run_dispatch_steps([{nl, Nl}, {qos, QoS}, {rap, Rap}], Msg, State);
+                        error ->
+                            dispatch(emqx_message:unset_flag(dup, Msg), State)
+                    end);
+        true ->
+            noreply(State)
+    end;
 
 %% Do nothing if the client has been disconnected.
 handle_info({timeout, Timer, retry_delivery}, State = #state{conn_pid = undefined, retry_timer = Timer}) ->

src/emqx_sm.erl

@@ -56,10 +56,15 @@ open_session(SessAttrs = #{clean_start := true, client_id := ClientId, conn_pid
                  end,
     emqx_sm_locker:trans(ClientId, CleanStart);
 
-open_session(SessAttrs = #{clean_start := false, client_id := ClientId, conn_pid := ConnPid}) ->
+open_session(SessAttrs = #{clean_start := false, 
+                           client_id := ClientId, 
+                           conn_pid := ConnPid, 
+                           max_inflight := MaxInflight,
+                           topic_alias_maximum := TopicAliasMaximum}) ->
     ResumeStart = fun(_) ->
                       case resume_session(ClientId, ConnPid) of
                           {ok, SPid} ->
+                              emqx_session:update_misc(SPid, #{max_inflight => MaxInflight, topic_alias_maximum => TopicAliasMaximum}),
                               {ok, SPid, true};
                           {error, not_found} ->
                               emqx_session_sup:start_session(SessAttrs)

test/emqx_mock_client.erl

@@ -51,7 +51,9 @@ handle_call({start_session, ClientPid, ClientId, Zone}, _From, State) ->
                conn_pid         => ClientPid,
                clean_start      => true,
                username         => undefined,
-               expiry_interval  => 0
+               expiry_interval  => 0,
+               max_inflight     => 0,
+               topic_alias_maximum => 0
              },
     {ok, SessPid} = emqx_sm:open_session(Attrs),
     {reply, {ok, SessPid},

test/emqx_mqtt_caps_SUITE.erl

@@ -44,7 +44,8 @@ t_get_set_caps(_) ->
     end,
     PubCaps = #{
         max_qos_allowed => ?QOS_2,
-        mqtt_retain_available => true
+        mqtt_retain_available => true,
+        max_topic_alias => 0
     },
     PubCaps = emqx_mqtt_caps:get_caps(zone, publish),
     NewPubCaps = PubCaps#{max_qos_allowed => ?QOS_1},

test/emqx_packet_SUITE.erl

@@ -46,7 +46,12 @@ packet_type_name(_) ->
 packet_validate(_) ->
     ?assertEqual(true, emqx_packet:validate(?SUBSCRIBE_PACKET(15, #{'Subscription-Identifier' => 1}, [{<<"topic">>, #{qos => ?QOS0}}]))),
     ?assertEqual(true, emqx_packet:validate(?UNSUBSCRIBE_PACKET(89, [<<"topic">>]))),
-    ?assertEqual(true, emqx_packet:validate(?CONNECT_PACKET(#mqtt_packet_connect{}))).
+    ?assertEqual(true, emqx_packet:validate(?CONNECT_PACKET(#mqtt_packet_connect{}))),
+    ?assertEqual(true, emqx_packet:validate(?CONNECT_PACKET(#mqtt_packet_connect{properties = #{'Receive-Maximum' => 1}}))),
+    case catch emqx_packet:validate(?CONNECT_PACKET(#mqtt_packet_connect{properties = #{'Receive-Maximum' => 0}})) of
+        {'EXIT', {protocol_error, _}} -> ?assertEqual(true, true);
+        true -> ?assertEqual(true, false)
+    end.
 
 packet_message(_) ->
     Pkt = #mqtt_packet{header = #mqtt_packet_header{type   = ?PUBLISH,

test/emqx_sm_SUITE.erl

@@ -25,7 +25,7 @@ t_open_close_session(_) ->
     emqx_ct_broker_helpers:run_setup_steps(),
     {ok, ClientPid} = emqx_mock_client:start_link(<<"client">>),
     Attrs = #{clean_start => true, client_id => <<"client">>, conn_pid => ClientPid,
-              zone => internal, username => <<"zhou">>, expiry_interval => 0},
+              zone => internal, username => <<"zhou">>, expiry_interval => 0, max_inflight => 0, topic_alias_maximum => 0},
     {ok, SPid} = emqx_sm:open_session(Attrs),
     [{<<"client">>, SPid}] = emqx_sm:lookup_session(<<"client">>),
     SPid = emqx_sm:lookup_session_pid(<<"client">>),