Fixed conflicts

etc/emqx.conf

@@ -779,7 +779,7 @@ listener.tcp.internal.acceptors = 4
 ## Maximum number of concurrent MQTT/TCP connections.
 ##
 ## Value: Number
-listener.tcp.internal.max_connections = 10240000
+listener.tcp.internal.max_connections = 1024000
 
 ## Maximum internal connections per second.
 ##

src/emqx_access_rule.erl

@@ -34,16 +34,21 @@
 -export([match/3]).
 
 -define(ALLOW_DENY(A), ((A =:= allow) orelse (A =:= deny))).
+-define(PUBSUB(A), ((A =:= subscribe) orelse (A =:= publish) orelse (A =:= pubsub))).
 
 %% @doc Compile Access Rule.
 compile({A, all}) when ?ALLOW_DENY(A) ->
     {A, all};
 
-compile({A, Who, Access, Topic}) when ?ALLOW_DENY(A), is_binary(Topic) ->
+compile({A, Who, Access, Topic}) when ?ALLOW_DENY(A), ?PUBSUB(Access), is_binary(Topic) ->
     {A, compile(who, Who), Access, [compile(topic, Topic)]};
 
-compile({A, Who, Access, TopicFilters}) when ?ALLOW_DENY(A) ->
-    {A, compile(who, Who), Access, [compile(topic, Topic) || Topic <- TopicFilters]}.
+compile({A, Who, Access, TopicFilters}) when ?ALLOW_DENY(A), ?PUBSUB(Access) ->
+    {A, compile(who, Who), Access, [compile(topic, Topic) || Topic <- TopicFilters]};
+
+compile(Rule) ->
+    emqx_logger:error("[ACCESS_RULE] Malformed rule: ~p", [Rule]),
+    {error, bad_rule}.
 
 compile(who, all) ->
     all;

src/emqx_acl_internal.erl

@@ -45,19 +45,27 @@ all_rules() ->
 
 -spec(init([File :: string()]) -> {ok, #{}}).
 init([File]) ->
-    ok = emqx_tables:new(?ACL_RULE_TAB, [set, public, {read_concurrency, true}]),
-    true = load_rules_from_file(File),
+    _ = emqx_tables:new(?ACL_RULE_TAB, [set, public, {read_concurrency, true}]),
+    ok = load_rules_from_file(File),
     {ok, #{acl_file => File}}.
 
 load_rules_from_file(AclFile) ->
-    {ok, Terms} = file:consult(AclFile),
-    Rules = [emqx_access_rule:compile(Term) || Term <- Terms],
-    lists:foreach(fun(PubSub) ->
-        ets:insert(?ACL_RULE_TAB, {PubSub,
-            lists:filter(fun(Rule) -> filter(PubSub, Rule) end, Rules)})
-        end, [publish, subscribe]),
-    ets:insert(?ACL_RULE_TAB, {all_rules, Terms}).
+    case file:consult(AclFile) of
+        {ok, Terms} ->
+            Rules = [emqx_access_rule:compile(Term) || Term <- Terms],
+            lists:foreach(fun(PubSub) ->
+                ets:insert(?ACL_RULE_TAB, {PubSub,
+                    lists:filter(fun(Rule) -> filter(PubSub, Rule) end, Rules)})
+                end, [publish, subscribe]),
+            ets:insert(?ACL_RULE_TAB, {all_rules, Terms}),
+            ok;
+        {error, Reason} ->
+            emqx_logger:error("[ACL_INTERNAL] Failed to read ~s: ~p", [AclFile, Reason]),
+            {error, Reason}
+    end.
 
+filter(_PubSub, {error, _}) ->
+    false;   
 filter(_PubSub, {allow, all}) ->
     true;
 filter(_PubSub, {deny, all}) ->
@@ -100,9 +108,11 @@ match(Credentials, Topic, [Rule|Rules]) ->
 -spec(reload_acl(state()) -> ok | {error, term()}).
 reload_acl(#{acl_file := AclFile}) ->
     case catch load_rules_from_file(AclFile) of
-        true ->
+        ok ->
             emqx_logger:info("Reload acl_file ~s successfully", [AclFile]),
             ok;
+        {error, Error} ->
+            {error, Error};
         {'EXIT', Error} ->
             {error, Error}
     end.

src/emqx_banned.erl

@@ -104,7 +104,7 @@ code_change(_OldVsn, State, _Extra) ->
 
 -ifdef(TEST).
 ensure_expiry_timer(State) ->
-    State#{expiry_timer := emqx_misc:start_timer(timer:seconds(2), expire)}.
+    State#{expiry_timer := emqx_misc:start_timer(timer:seconds(1), expire)}.
 -else.
 ensure_expiry_timer(State) ->
     State#{expiry_timer := emqx_misc:start_timer(timer:minutes(1), expire)}.

src/emqx_protocol.erl

@@ -620,14 +620,12 @@ maybe_use_username_as_clientid(ClientId, undefined, _PState) ->
     ClientId;
 maybe_use_username_as_clientid(ClientId, Username, #pstate{zone = Zone}) ->
     case emqx_zone:get_env(Zone, use_username_as_clientid, false) of
-        true ->
-            Username;
-        false ->
-            ClientId
+        true -> Username;
+        false -> ClientId
     end.
 
 %%------------------------------------------------------------------------------
-%% Assign a clientid
+%% Assign a clientId
 
 maybe_assign_client_id(PState = #pstate{client_id = <<>>, ack_props = AckProps}) ->
     ClientId = emqx_guid:to_base62(emqx_guid:gen()),
@@ -651,41 +649,30 @@ try_open_session(PState = #pstate{zone        = Zone,
         clean_start => CleanStart,
         will_msg    => WillMsg
     },
-    SessAttrs1 = lists:foldl(fun set_session_attrs/2, SessAttrs, [{max_inflight, PState}, {expiry_interval, PState}, {topic_alias_maximum, PState}]),
+    SessAttrs1 = lists:foldl(fun set_session_attrs/2, SessAttrs, [{max_inflight, PState}, {expiry_interval, 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 ->
-                                   get_property('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 ->
-                                   get_property('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 ->
-                                        get_property('Topic-Alias-Maximum', ConnProps, 0);
-                                    true ->
-                                        emqx_zone:get_env(Zone, max_topic_alias, 0)
-                                  end, SessAttrs);
-set_session_attrs({_, #pstate{}}, SessAttrs) ->
-    SessAttrs.
+set_session_attrs({max_inflight, #pstate{proto_ver = ?MQTT_PROTO_V5, conn_props = ConnProps}}, SessAttrs) ->
+    maps:put(max_inflight, get_property('Receive-Maximum', ConnProps, 65535), SessAttrs);
+
+set_session_attrs({max_inflight, #pstate{zone = Zone}}, SessAttrs) ->
+    maps:put(max_inflight, emqx_zone:get_env(Zone, max_inflight, 65535), SessAttrs);
 
+set_session_attrs({expiry_interval, #pstate{proto_ver = ?MQTT_PROTO_V5, conn_props = ConnProps}}, SessAttrs) ->
+    maps:put(expiry_interval, get_property('Session-Expiry-Interval', ConnProps, 0), SessAttrs);
+
+set_session_attrs({expiry_interval, #pstate{zone = Zone, clean_start = CleanStart}}, SessAttrs) ->
+    maps:put(expiry_interval, case CleanStart of
+                                  true -> 0;
+                                  false -> emqx_zone:get_env(Zone, session_expiry_interval, 16#ffffffff)
+                              end, SessAttrs);
+
+set_session_attrs(_, SessAttrs) ->
+    SessAttrs.
 
 authenticate(Credentials, Password) ->
     case emqx_access_control:authenticate(Credentials, Password) of

src/emqx_session.erl

@@ -147,8 +147,6 @@
           %% Created at
           created_at :: erlang:timestamp(),
 
-          topic_alias_maximum :: pos_integer(),
-
           will_msg :: emqx:message(),
 
           will_delay_timer :: reference() | undefined
@@ -341,7 +339,6 @@ init([Parent, #{zone                := Zone,
                 clean_start         := CleanStart,
                 expiry_interval     := ExpiryInterval,
                 max_inflight        := MaxInflight,
-                topic_alias_maximum := TopicAliasMaximum,
                 will_msg            := WillMsg}]) ->
     emqx_logger:set_metadata_client_id(ClientId),
     process_flag(trap_exit, true),
@@ -367,7 +364,6 @@ init([Parent, #{zone                := Zone,
                    deliver_stats       = 0,
                    enqueue_stats       = 0,
                    created_at          = os:timestamp(),
-                   topic_alias_maximum = TopicAliasMaximum,
                    will_msg            = WillMsg
                   },
     ok = emqx_sm:register_session(ClientId, self()),
@@ -519,22 +515,23 @@ handle_cast({pubcomp, PacketId, _ReasonCode}, State = #state{inflight = Inflight
     end;
 
 %% RESUME:
-handle_cast({resume, #{conn_pid            := ConnPid,
-                       will_msg            := WillMsg,
-                       expiry_interval     := SessionExpiryInterval,
-                       max_inflight        := MaxInflight,
-                       topic_alias_maximum := TopicAliasMaximum}}, State = #state{client_id        = ClientId,
-                                                                                  conn_pid         = OldConnPid,
-                                                                                  clean_start      = CleanStart,
-                                                                                  retry_timer      = RetryTimer,
-                                                                                  await_rel_timer  = AwaitTimer,
-                                                                                  expiry_timer     = ExpireTimer,
-                                                                                  will_delay_timer = WillDelayTimer}) ->
+handle_cast({resume, #{conn_pid        := ConnPid,
+                       will_msg        := WillMsg,
+                       expiry_interval := ExpiryInterval,
+                       max_inflight    := MaxInflight}},
+            State = #state{client_id        = ClientId,
+                           conn_pid         = OldConnPid,
+                           clean_start      = CleanStart,
+                           retry_timer      = RetryTimer,
+                           await_rel_timer  = AwaitTimer,
+                           expiry_timer     = ExpireTimer,
+                           will_delay_timer = WillDelayTimer}) ->
 
     ?LOG(info, "Resumed by connection ~p ", [ConnPid], State),
 
     %% Cancel Timers
-    lists:foreach(fun emqx_misc:cancel_timer/1, [RetryTimer, AwaitTimer, ExpireTimer, WillDelayTimer]),
+    lists:foreach(fun emqx_misc:cancel_timer/1,
+                  [RetryTimer, AwaitTimer, ExpireTimer, WillDelayTimer]),
 
     case kick(ClientId, OldConnPid, ConnPid) of
         ok -> ?LOG(warning, "Connection ~p kickout ~p", [ConnPid, OldConnPid], State);
@@ -543,19 +540,18 @@ handle_cast({resume, #{conn_pid            := ConnPid,
 
     true = link(ConnPid),
 
-    State1 = State#state{conn_pid            = ConnPid,
-                         binding             = binding(ConnPid),
-                         old_conn_pid        = OldConnPid,
-                         clean_start         = false,
-                         retry_timer         = undefined,
-                         awaiting_rel        = #{},
-                         await_rel_timer     = undefined,
-                         expiry_timer        = undefined,
-                         expiry_interval     = SessionExpiryInterval,
-                         inflight            = emqx_inflight:update_size(MaxInflight, State#state.inflight),
-                         topic_alias_maximum = TopicAliasMaximum,
-                         will_delay_timer    = undefined,
-                         will_msg            = WillMsg},
+    State1 = State#state{conn_pid         = ConnPid,
+                         binding          = binding(ConnPid),
+                         old_conn_pid     = OldConnPid,
+                         clean_start      = false,
+                         retry_timer      = undefined,
+                         awaiting_rel     = #{},
+                         await_rel_timer  = undefined,
+                         expiry_timer     = undefined,
+                         expiry_interval  = ExpiryInterval,
+                         inflight         = emqx_inflight:update_size(MaxInflight, State#state.inflight),
+                         will_delay_timer = undefined,
+                         will_msg         = WillMsg},
 
     %% Clean Session: true -> false???
     CleanStart andalso emqx_sm:set_session_attrs(ClientId, attrs(State1)),
@@ -574,9 +570,10 @@ handle_cast(Msg, State) ->
 
 %% Batch dispatch
 handle_info({dispatch, Topic, Msgs}, State) when is_list(Msgs) ->
-    {noreply, lists:foldl(fun(Msg, NewState) ->
-                                  element(2, handle_info({dispatch, Topic, Msg}, NewState))
-                          end, State, Msgs)};
+    noreply(lists:foldl(
+              fun(Msg, St) ->
+                  element(2, handle_info({dispatch, Topic, Msg}, St))
+              end, State, Msgs));
 
 %% Dispatch message
 handle_info({dispatch, Topic, Msg = #message{}}, State) ->
@@ -585,12 +582,11 @@ handle_info({dispatch, Topic, Msg = #message{}}, State) ->
             %% Require ack, but we do not have connection
             %% negative ack the message so it can try the next subscriber in the group
             ok = emqx_shared_sub:nack_no_connection(Msg),
-            noreply(State);
+            {noreply, State};
         false ->
-            handle_dispatch(Topic, Msg, State)
+            noreply(handle_dispatch(Topic, Msg, State))
     end;
 
-
 %% Do nothing if the client has been disconnected.
 handle_info({timeout, Timer, retry_delivery}, State = #state{conn_pid = undefined, retry_timer = Timer}) ->
     noreply(State#state{retry_timer = undefined});
@@ -678,23 +674,14 @@ maybe_shutdown(Pid, Reason) ->
 has_connection(#state{conn_pid = Pid}) ->
     is_pid(Pid) andalso is_process_alive(Pid).
 
-handle_dispatch(Topic, Msg = #message{headers = Headers},
-                State = #state{subscriptions = SubMap,
-                               topic_alias_maximum = TopicAliasMaximum
-                              }) ->
-    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)
+handle_dispatch(Topic, Msg, State = #state{subscriptions = SubMap}) ->
+    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.
 
 suback(_From, undefined, _ReasonCodes) ->

test/emqx_access_SUITE.erl

@@ -355,7 +355,8 @@ compile_rule(_) ->
     {deny, all, subscribe, [ [<<"$SYS">>, '#'], ['#'] ]} =
         compile({deny, all, subscribe, ["$SYS/#", "#"]}),
     {allow, all} = compile({allow, all}),
-    {deny, all} = compile({deny, all}).
+    {deny, all} = compile({deny, all}),
+    {error, bad_rule} = compile({test, malformed}).
 
 match_rule(_) ->
     User = #{client_id => <<"testClient">>, username => <<"TestUser">>, peername => {{127,0,0,1}, 2948}},