Merge pull request #6174 from zmstone/main-v4.4-alpha.1

sync v4.3 to v4.4

apps/emqx_rule_engine/src/emqx_rule_engine.app.src

@@ -1,6 +1,6 @@
 {application, emqx_rule_engine,
  [{description, "EMQ X Rule Engine"},
-  {vsn, "4.3.6"}, % strict semver, bump manually!
+  {vsn, "4.4.0"}, % strict semver, bump manually!
   {modules, []},
   {registered, [emqx_rule_engine_sup, emqx_rule_registry]},
   {applications, [kernel,stdlib,rulesql,getopt]},

apps/emqx_stomp/src/emqx_stomp.appup.src

@@ -5,7 +5,8 @@
      {load_module,emqx_stomp_protocol,brutal_purge,soft_purge,[]},
      {load_module,emqx_stomp_connection,brutal_purge,soft_purge,[]}]},
    {"4.3.0",
-    [{restart_application,emqx_stomp}]},
+    [{restart_application,emqx_stomp},
+     {apply,{emqx_stomp,force_clear_after_app_stoped,[]}}]},
    {<<".*">>,[]}],
   [{"4.3.2",[{load_module,emqx_stomp_protocol,brutal_purge,soft_purge,[]}]},
    {"4.3.1",[

apps/emqx_stomp/src/emqx_stomp.erl

@@ -33,6 +33,8 @@
         , stop_listener/3
         ]).
 
+-export([force_clear_after_app_stoped/0]).
+
 -export([init/1]).
 
 -define(APP, ?MODULE).
@@ -52,6 +54,18 @@ start(_StartType, _StartArgs) ->
 stop(_State) ->
     stop_listeners().
 
+force_clear_after_app_stoped() ->
+    lists:foreach(fun({Name = {ProtoName, _}, _}) ->
+        case is_stomp_listener(ProtoName) of
+            true -> esockd:close(Name);
+            _ -> ok
+        end
+    end, esockd:listeners()).
+
+is_stomp_listener('stomp:tcp') -> true;
+is_stomp_listener('stomp:ssl') -> true;
+is_stomp_listener(_) -> false.
+
 %%--------------------------------------------------------------------
 %% Supervisor callbacks
 %%--------------------------------------------------------------------

include/emqx_release.hrl

@@ -29,7 +29,7 @@
 
 -ifndef(EMQX_ENTERPRISE).
 
--define(EMQX_RELEASE, {opensource, "4.4.0"}).
+-define(EMQX_RELEASE, {opensource, "4.4.0-alpha.1"}).
 
 -else.
 

lib-ce/emqx_dashboard/src/emqx_dashboard.app.src

@@ -1,6 +1,6 @@
 {application, emqx_dashboard,
  [{description, "EMQ X Web Dashboard"},
-  {vsn, "4.3.6"}, % strict semver, bump manually!
+  {vsn, "4.3.7"}, % strict semver, bump manually!
   {modules, []},
   {registered, [emqx_dashboard_sup]},
   {applications, [kernel,stdlib,mnesia,minirest]},

pkg-vsn.sh

@@ -15,9 +15,21 @@ fi
 ## emqx_release.hrl is the single source of truth for release version
 RELEASE="$(grep -E "define.+EMQX_RELEASE.+${EDITION}" include/emqx_release.hrl | cut -d '"' -f2)"
 
-## git commit hash is added as suffix in case the git tag and release version is not an exact match
+git_exact_vsn() {
-if [ -d .git ] && ! git describe --tags --match "[e|v]${RELEASE}" --exact >/dev/null 2>&1; then
+    local tag
+    tag="$(git describe --tags --match "[e|v]*" --exact 2>/dev/null)"
+    echo "$tag" | sed 's/^[v|e]//g'
+}
+
+GIT_EXACT_VSN="$(git_exact_vsn)"
+if [ "$GIT_EXACT_VSN" != '' ]; then
+    if [ "$GIT_EXACT_VSN" != "$RELEASE" ]; then
+        echo "ERROR: Tagged $GIT_EXACT_VSN, but $RELEASE in include/emqx_release.hrl" 1>&2
+        exit 1
+    fi
+    SUFFIX=''
+else
     SUFFIX="-$(git rev-parse HEAD | cut -b1-8)"
 fi
 
-echo "${RELEASE}${SUFFIX:-}"
+echo "${RELEASE}${SUFFIX}"

rebar.config

@@ -37,7 +37,7 @@
 
 {deps,
     [ {gpb, "4.11.2"} %% gpb only used to build, but not for release, pin it here to avoid fetching a wrong version due to rebar plugins scattered in all the deps
-    , {ehttpc, {git, "https://github.com/emqx/ehttpc", {tag, "0.1.10"}}}
+    , {ehttpc, {git, "https://github.com/emqx/ehttpc", {tag, "0.1.12"}}}
     , {eredis_cluster, {git, "https://github.com/emqx/eredis_cluster", {tag, "0.6.5"}}}
     , {gproc, {git, "https://github.com/uwiger/gproc", {tag, "0.8.0"}}}
     , {jiffy, {git, "https://github.com/emqx/jiffy", {tag, "1.0.5"}}}

src/emqx.appup.src

@@ -1,6 +1,8 @@
 %% -*- mode: erlang -*-
 {VSN,
-  [{"4.3.10",[{load_module,emqx_connection,brutal_purge,soft_purge,[]}]},
+  [{"4.3.10",
+    [{load_module,emqx_app,brutal_purge,soft_purge,[]},
+     {load_module,emqx_connection,brutal_purge,soft_purge,[]}]},
    {"4.3.9",
     [{load_module,emqx_connection,brutal_purge,soft_purge,[]},
      {load_module,emqx_channel,brutal_purge,soft_purge,[]},
@@ -155,7 +157,9 @@
      {load_module,emqx_rpc,brutal_purge,soft_purge,[]},
      {load_module,emqx_app,brutal_purge,soft_purge,[]}]},
    {<<".*">>,[]}],
-  [{"4.3.10",[{load_module,emqx_connection,brutal_purge,soft_purge,[]}]},
+  [{"4.3.10",
+    [{load_module,emqx_app,brutal_purge,soft_purge,[]},
+     {load_module,emqx_connection,brutal_purge,soft_purge,[]}]},
    {"4.3.9",
     [{load_module,emqx_connection,brutal_purge,soft_purge,[]},
      {load_module,emqx_channel,brutal_purge,soft_purge,[]},

src/emqx_app.erl

@@ -42,10 +42,13 @@ start(_Type, _Args) ->
     ekka:start(),
     {ok, Sup} = emqx_sup:start_link(),
     ok = start_autocluster(),
+    %% We need to make sure that emqx's listeners start before plugins
+    %% and modules. Since if the emqx-conf module/plugin is enabled, it will
+    %% try to start or update the listeners with the latest configuration
+    emqx_boot:is_enabled(listeners) andalso (ok = emqx_listeners:start()),
     ok = emqx_plugins:init(),
     _ = emqx_plugins:load(),
     _ = start_ce_modules(),
-    emqx_boot:is_enabled(listeners) andalso (ok = emqx_listeners:start()),
     register(emqx, self()),
     ok = emqx_alarm_handler:load(),
     print_vsn(),

test/emqx_cm_SUITE.erl

@@ -183,7 +183,8 @@ t_open_session_race_condition(_) ->
 
     exit(Winner, kill),
     receive {'DOWN', _, process, Winner, _} -> ok end,
-    ignored = gen_server:call(emqx_cm, ignore, infinity), %% sync
+    ignored = gen_server:call(?CM, ignore, infinity), %% sync
+    ok = flush_emqx_pool(),
     ?assertEqual([], emqx_cm:lookup_channels(ClientId)).
 
 t_kick_session_discard_normal(_) ->
@@ -260,6 +261,7 @@ test_kick_session(Action, Reason) ->
             ?assertEqual(Reason, ?WAIT({'DOWN', _, process, Pid1, R}, 2_000, R)),
             ?assertEqual(Reason, ?WAIT({'DOWN', _, process, Pid2, R}, 2_000, R))
     end,
+    ignored = gen_server:call(?CM, ignore, infinity), %% sync
     ok = flush_emqx_pool(),
     ?assertEqual([], emqx_cm:lookup_channels(ClientId)).
 
@@ -271,10 +273,11 @@ test_kick_session(Action, Reason) ->
 %% The number of tasks should be large enough to ensure all workers have
 %% the chance to work on at least one of the tasks.
 flush_emqx_pool() ->
+    Ref = make_ref(),
     Self = self(),
     L = lists:seq(1, 1000),
-    lists:foreach(fun(I) -> emqx_pool:async_submit(fun() -> Self ! {done, I} end, []) end, L),
+    lists:foreach(fun(I) -> emqx_pool:async_submit(fun() -> Self ! {done, I, Ref} end, []) end, L),
-    lists:foreach(fun(I) -> receive {done, I} -> ok end end, L).
+    lists:foreach(fun(I) -> receive {done, I, Ref} -> ok end end, L).
 
 t_discard_session_race(_) ->
     ClientId = rand_client_id(),