Merge remote-tracking branch 'origin/main-v4.3' into main-v4.4

priv/emqx.schema

@@ -264,6 +264,28 @@ end}.
 {validator, "range4ports", "must be 1024 to 134217727",
  fun(X) -> X >= 1024 andalso X =< 134217727 end}.
 
+{validator, "range:0-2", "must be 0 to 2",
+ fun(X) -> X >= 0 andalso X =< 2 end}.
+
+{validator, "range:0-128", "must be 0 to 128",
+  fun(X) -> X >= 0 andalso X =< 128 end}.
+
+{validator, "range:0-65535", "must be 0 to 65535",
+ fun(X) -> X >= 0 andalso X =< 65535 end}.
+
+{validator, "range:1-65535", "must be 1 to 65535",
+  fun(X) -> X >= 1 andalso X =< 65535 end}.
+
+{validator, "range:1-9", "must be 1 to 9",
+  fun(X) -> X >= 1 andalso X =< 9 end}.
+
+{validator, "range:8-15", "must be 8 to 15",
+    fun(X) -> X >= 8 andalso X =< 15 end}.
+
+{validator, "range:0-1024", "must be 0 to 1024",
+  fun(X) -> X >= 0 andalso X =< 1024 end}.
+
+
 %% @doc http://www.erlang.org/doc/man/erl.html#%2bzdbbl
 {mapping, "node.dist_buffer_size", "vm_args.+zdbbl", [
   {datatype, bytesize},
@@ -299,10 +321,10 @@ end}.
   {default, 1000},
   {datatype, integer},
   hidden,
-  {validators, ["positive_integer"]}
+  {validators, ["range:0-inf"]}
 ]}.
 
-{validator, "positive_integer", "must be a positive integer",
+{validator, "range:0-inf", "must be a non neg_integer",
   fun(X) -> X >= 0 end}.
 
 %% Note: OTP R15 and earlier uses -env ERL_MAX_ETS_TABLES,
@@ -836,7 +858,8 @@ end}.
 %% @doc Set the Maximum topic levels.
 {mapping, "mqtt.max_topic_levels", "emqx.max_topic_levels", [
   {default, 128},
-  {datatype, integer}
+  {datatype, integer},
+  {validators, ["range:0-inf"]}
 ]}.
 
 %% @doc Set the Maximum QoS allowed.
@@ -849,7 +872,8 @@ end}.
 %% @doc Set the Maximum Topic Alias.
 {mapping, "mqtt.max_topic_alias", "emqx.max_topic_alias", [
   {default, 65535},
-  {datatype, integer}
+  {datatype, integer},
+  {validators, ["range:0-65535"]}
 ]}.
 
 %% @doc Whether the server supports MQTT retained messages.
@@ -946,7 +970,8 @@ end}.
 
 %% @doc Set the Maximum topic levels.
 {mapping, "zone.$name.max_topic_levels", "emqx.zones", [
-  {datatype, integer}
+  {datatype, integer},
+  {validators, ["range:0-128"]}
 ]}.
 
 %% @doc Set the Maximum QoS allowed.

scripts/update-appup.sh

@@ -16,12 +16,15 @@ cd -P -- "$(dirname -- "${BASH_SOURCE[0]}")/.."
 PROFILE="${1:-}"
 case "$PROFILE" in
     emqx-ee)
+        GIT_REPO='emqx/emqx-enterprise.git'
         TAG_PREFIX='e'
         ;;
     emqx)
+        GIT_REPO='emqx/emqx.git'
         TAG_PREFIX='v'
         ;;
     emqx-edge)
+        GIT_REPO='emqx/emqx.git'
         TAG_PREFIX='v'
         ;;
     *)
@@ -36,7 +39,7 @@ esac
 ##  e4.3.11
 ##  rel-v4.4.3
 ##  rel-e4.4.3
-PREV_TAG="$(git describe --tag --abbrev=0 --match "[${TAG_PREFIX}|rel-]*" --exclude '*rc*' --exclude '*alpha*' --exclude '*beta*')"
+PREV_TAG="${PREV_TAG:-$(git describe --tag --abbrev=0 --match "[${TAG_PREFIX}|rel-]*" --exclude '*rc*' --exclude '*alpha*' --exclude '*beta*')}"
 
 shift 1
 # bash 3.2 treat empty array as unbound, so we can't use 'ESCRIPT_ARGS=()' here,
@@ -75,16 +78,23 @@ if [ "${SKIP_BUILD:-}" != 'yes' ]; then
     make "${PROFILE}"
 fi
 
-PREV_DIR_BASE="/tmp/emqx-appup-build"
+PREV_DIR_BASE="/tmp/_w"
 mkdir -p "${PREV_DIR_BASE}"
-if [ ! -d "${PREV_TAG}" ]; then
+if [ ! -d "${PREV_DIR_BASE}/${PREV_TAG}" ]; then
     cp -R . "${PREV_DIR_BASE}/${PREV_TAG}"
+    # always 'yes' in CI
+    NEW_COPY='yes'
+else
+    NEW_COPY='no'
 fi
 
 pushd "${PREV_DIR_BASE}/${PREV_TAG}"
-git reset --hard
+if [ "$NEW_COPY" = 'no' ]; then
+    REMOTE="$(git remote -v | grep "${GIT_REPO}" | head -1 | awk '{print $1}')"
+    git fetch "$REMOTE"
+fi
+git clean -fdx
 git checkout "${PREV_TAG}"
-make clean-all
 make "$PROFILE"
 popd
 

src/emqx_cm_locker.erl

@@ -28,6 +28,11 @@
         , unlock/1
         ]).
 
+%% for testing
+-ifdef(TEST).
+-export([strategy/0]).
+-endif.
+
 -spec(start_link() -> startlink_ret()).
 start_link() ->
     ekka_locker:start_link(?MODULE).
@@ -63,4 +68,3 @@ unlock(ClientId) ->
 -spec(strategy() -> local | leader | quorum | all).
 strategy() ->
     emqx:get_env(session_locking_strategy, quorum).
-

src/emqx_flapping.erl

@@ -38,6 +38,10 @@
         , code_change/3
         ]).
 
+%% for testing
+-ifdef(TEST).
+-export([get_policy/0]).
+-endif.
 %% Tab
 -define(FLAPPING_TAB, ?MODULE).
 %% Default Policy
@@ -160,4 +164,4 @@ code_change(_OldVsn, State, _Extra) ->
 ensure_timer(State) ->
     Timeout = maps:get(duration, get_policy()),
     TRef = emqx_misc:start_timer(Timeout, expired_detecting),
-    State#{expired_timer => TRef}.
+    State#{expired_timer => TRef}.

src/emqx_limiter.erl

@@ -23,7 +23,8 @@
         , init/4 %% XXX: Compatible with before 4.2 version
         , info/1
         , check/2
-        , update_overall_limiter/4
+        , update_overall_limiter/3
+        , delete_overall_limiter/1
         ]).
 
 -record(limiter, {
@@ -154,14 +155,18 @@ is_message_limiter(conn_messages_routing) -> true;
 is_message_limiter(overall_messages_routing) -> true;
 is_message_limiter(_) -> false.
 
-update_overall_limiter(Zone, Name, Capacity, Interval) ->
+update_overall_limiter(Zone, Capacity, Interval) ->
-    case is_overall_limiter(Name) of
+    try
-        false -> false;
+        esockd_limiter:update({Zone, overall_messages_routing}, Capacity, Interval),
-        _ ->
+        true
-            try
+    catch _:_:_ ->
-                esockd_limiter:update({Zone, Name}, Capacity, Interval),
+        false
-                true
+    end.
-            catch _:_:_ ->
+
-                    false
+delete_overall_limiter(Zone) ->
-            end
+    try
+        esockd_limiter:delete({Zone, overall_messages_routing}),
+        true
+    catch _:_:_ ->
+        false
     end.

src/emqx_listeners.erl

@@ -29,6 +29,7 @@
 -export([ start_listener/1
         , start_listener/3
         , stop_listener/1
+        , update_listeners_env/2
         , restart_listener/1
         , restart_listener/3
         ]).
@@ -187,6 +188,20 @@ with_port(Port, Opts = #{socket_opts := SocketOption}) when is_integer(Port) ->
 with_port({Addr, Port}, Opts = #{socket_opts := SocketOption}) ->
     Opts#{socket_opts => [{ip, Addr}, {port, Port}| SocketOption]}.
 
+update_listeners_env(Action, NewConf = #{name := NewName, proto := NewProto}) ->
+    Listener = emqx:get_env(listeners, []),
+    Listener1 = lists:filter(
+        fun(#{name := Name, proto := Proto}) ->
+            not (Name =:= NewName andalso Proto =:= NewProto)
+        end, Listener),
+    Listener2 =
+        case Action of
+            update -> [NewConf | Listener1];
+            delete -> Listener1
+        end,
+    application:set_env(emqx, listeners, Listener2),
+    ok.
+
 %% @doc Restart all listeners
 -spec(restart() -> ok).
 restart() ->

src/emqx_shared_sub.erl

@@ -47,7 +47,7 @@
         ]).
 
 %% for testing
--export([subscribers/2]).
+-export([subscribers/2, ack_enabled/0]).
 
 %% gen_server callbacks
 -export([ init/1