Merge remote-tracking branch 'origin/dev/v4.3.0' into merge-dev-v4.3.0

apps/emqx_rule_engine/src/emqx_rule_funcs.erl

@@ -706,52 +706,10 @@ map_get(Key, Map) ->
     map_get(Key, Map, undefined).
 
 map_get(Key, Map, Default) ->
-    case maps:find(Key, Map) of
-        {ok, Val} -> Val;
-        error when is_atom(Key) ->
-            %% the map may have an equivalent binary-form key
-            BinKey = emqx_rule_utils:bin(Key),
-            case maps:find(BinKey, Map) of
-                {ok, Val} -> Val;
-                error -> Default
-            end;
-        error when is_binary(Key) ->
-            try %% the map may have an equivalent atom-form key
-                AtomKey = list_to_existing_atom(binary_to_list(Key)),
-                case maps:find(AtomKey, Map) of
-                    {ok, Val} -> Val;
-                    error -> Default
-                end
-            catch error:badarg ->
-                Default
-            end;
-        error ->
-            Default
-    end.
+    emqx_rule_maps:nested_get(map_path(Key), Map, Default).
 
 map_put(Key, Val, Map) ->
-    case maps:find(Key, Map) of
-        {ok, _} -> maps:put(Key, Val, Map);
-        error when is_atom(Key) ->
-            %% the map may have an equivalent binary-form key
-            BinKey = emqx_rule_utils:bin(Key),
-            case maps:find(BinKey, Map) of
-                {ok, _} -> maps:put(BinKey, Val, Map);
-                error -> maps:put(Key, Val, Map)
-            end;
-        error when is_binary(Key) ->
-            try %% the map may have an equivalent atom-form key
-                AtomKey = list_to_existing_atom(binary_to_list(Key)),
-                case maps:find(AtomKey, Map) of
-                    {ok, _} -> maps:put(AtomKey, Val, Map);
-                    error -> maps:put(Key, Val, Map)
-                end
-            catch error:badarg ->
-                maps:put(Key, Val, Map)
-            end;
-        error ->
-            maps:put(Key, Val, Map)
-    end.
+    emqx_rule_maps:nested_put(map_path(Key), Val, Map).
 
 mget(Key, Map) ->
     mget(Key, Map, undefined).

apps/emqx_rule_engine/test/emqx_rule_engine_SUITE.erl

@@ -692,6 +692,48 @@ t_update_rule(_Config) ->
     ?assertEqual(not_found, emqx_rule_registry:get_rule(<<"an_existing_rule">>)),
     ok.
 
+t_disable_rule(_Config) ->
+    ets:new(simpile_action_2, [named_table, set, public]),
+    ets:insert(simpile_action_2, {created, 0}),
+    ets:insert(simpile_action_2, {destroyed, 0}),
+    Now = erlang:timestamp(),
+    emqx_rule_registry:add_action(
+        #action{name = 'simpile_action_2', app = ?APP,
+                module = ?MODULE,
+                on_create = simpile_action_2_create,
+                on_destroy = simpile_action_2_destroy,
+                types=[], params_spec = #{},
+                title = #{en => <<"Simple Action">>},
+                description = #{en => <<"Simple Action">>}}),
+    {ok, #rule{actions = [#action_instance{id = ActInsId0}]}} = emqx_rule_engine:create_rule(
+        #{id => <<"simple_rule_2">>,
+          rawsql => <<"select * from \"t/#\"">>,
+          actions => [#{name => 'simpile_action_2', args => #{}}]
+        }),
+    [{_, CAt}] = ets:lookup(simpile_action_2, created),
+    ?assert(CAt > Now),
+    [{_, DAt}] = ets:lookup(simpile_action_2, destroyed),
+    ?assert(DAt < Now),
+
+    %% disable the rule and verify the old action instances has been cleared
+    Now2 = erlang:timestamp(),
+    emqx_rule_engine:update_rule(#{ id => <<"simple_rule_2">>,
+                                    enabled => false}),
+    [{_, CAt2}] = ets:lookup(simpile_action_2, created),
+    ?assert(CAt2 < Now2),
+    [{_, DAt2}] = ets:lookup(simpile_action_2, destroyed),
+    ?assert(DAt2 > Now2),
+
+    %% enable the rule again and verify the action instances has been created
+    Now3 = erlang:timestamp(),
+    emqx_rule_engine:update_rule(#{ id => <<"simple_rule_2">>,
+                                    enabled => true}),
+    [{_, CAt3}] = ets:lookup(simpile_action_2, created),
+    ?assert(CAt3 > Now3),
+    [{_, DAt3}] = ets:lookup(simpile_action_2, destroyed),
+    ?assert(DAt3 < Now3),
+    ok = emqx_rule_engine:delete_rule(<<"simple_rule_2">>).
+
 t_get_rules_for(_Config) ->
     Len0 = length(emqx_rule_registry:get_rules_for(<<"simple/topic">>)),
     ok = emqx_rule_registry:add_rules(
@@ -2207,6 +2249,14 @@ crash_action(_Id, _Params) ->
         error(crash)
     end.
 
+simpile_action_2_create(_Id, _Params) ->
+    ets:insert(simpile_action_2, {created, erlang:timestamp()}),
+    fun(_Data, _Envs) -> ok end.
+
+simpile_action_2_destroy(_Id, _Params) ->
+    ets:insert(simpile_action_2, {destroyed, erlang:timestamp()}),
+    fun(_Data, _Envs) -> ok end.
+
 init_plus_by_one_action() ->
     ets:new(plus_by_one_action, [named_table, set, public]),
     ets:insert(plus_by_one_action, {num, 0}).

apps/emqx_rule_engine/test/emqx_rule_funcs_SUITE.erl

@@ -523,10 +523,16 @@ t_contains(_) ->
 
 t_map_get(_) ->
     ?assertEqual(1, apply_func(map_get, [<<"a">>, #{a => 1}])),
+    ?assertEqual(undefined, apply_func(map_get, [<<"a">>, #{}])),
+    ?assertEqual(1, apply_func(map_get, [<<"a.b">>, #{a => #{b => 1}}])),
+    ?assertEqual(undefined, apply_func(map_get, [<<"a.c">>, #{a => #{b => 1}}])),
     ?assertEqual(undefined, apply_func(map_get, [<<"a">>, #{}])).
 
 t_map_put(_) ->
     ?assertEqual(#{<<"a">> => 1}, apply_func(map_put, [<<"a">>, 1, #{}])),
+    ?assertEqual(#{a => 2}, apply_func(map_put, [<<"a">>, 2, #{a => 1}])),
+    ?assertEqual(#{<<"a">> => #{<<"b">> => 1}}, apply_func(map_put, [<<"a.b">>, 1, #{}])),
+    ?assertEqual(#{a => #{b => 1, <<"c">> => 1}}, apply_func(map_put, [<<"a.c">>, 1, #{a => #{b => 1}}])),
     ?assertEqual(#{a => 2}, apply_func(map_put, [<<"a">>, 2, #{a => 1}])).
 
 t_mget(_) ->

apps/emqx_telemetry/src/emqx_telemetry.erl

@@ -157,7 +157,12 @@ init([Opts]) ->
                  [#telemetry{uuid = UUID, enabled = Enabled} | _] ->
                      State#state{enabled = Enabled, uuid = UUID}
              end,
-    {ok, ensure_report_timer(NState), {continue, first_report}}.
+    case official_version(emqx_version()) of
+        true ->
+            {ok, ensure_report_timer(NState), {continue, first_report}};
+        false ->
+            {ok, NState#state{enabled = false}}
+    end. 
 
 handle_call(enable, _From, State = #state{uuid = UUID}) ->
     mnesia:dirty_write(?TELEMETRY, #telemetry{id = ?UNIQUE_ID,
@@ -217,6 +222,14 @@ code_change(_OldVsn, State, _Extra) ->
 %% Internal functions
 %%------------------------------------------------------------------------------
 
+official_version(Version) ->
+    case re:run(Version,
+                "^\\d+\\.\\d+(?:(-(?:alpha|beta|rc)\\.[1-9][0-9]*)|\\.\\d+)$",
+                [{capture, none}]) of
+        match -> true;
+        nomatch -> false
+    end.
+
 ensure_report_timer(State = #state{report_interval = ReportInterval}) ->
     State#state{timer = emqx_misc:start_timer(ReportInterval, time_to_report_telemetry_data)}.
 

apps/emqx_telemetry/test/emqx_telemetry_SUITE.erl

@@ -46,6 +46,20 @@ t_uuid(_) ->
     {ok, UUID3} = emqx_telemetry:get_uuid(),
     ?assertEqual(UUID2, UUID3).
 
+t_official_version(_) ->
+    true = emqx_telemetry:official_version("0.0.0"),
+    true = emqx_telemetry:official_version("1.1.1"),
+    true = emqx_telemetry:official_version("10.10.10"),
+    false = emqx_telemetry:official_version("0.0.0.0"),
+    false = emqx_telemetry:official_version("1.1.a"),
+    true = emqx_telemetry:official_version("0.0-alpha.1"),
+    true = emqx_telemetry:official_version("1.1-alpha.1"),
+    true = emqx_telemetry:official_version("10.10-alpha.10"),
+    false = emqx_telemetry:official_version("1.1-alpha.0"),
+    true = emqx_telemetry:official_version("1.1-beta.1"),
+    true = emqx_telemetry:official_version("1.1-rc.1"),
+    false = emqx_telemetry:official_version("1.1-alpha.a").
+
 t_get_telemetry(_) ->
     {ok, TelemetryData} = emqx_telemetry:get_telemetry(),
     OTPVersion = bin(erlang:system_info(otp_release)),

sync-apps.sh

@@ -25,7 +25,7 @@ apps=(
 "emqx_psk_file"
 "emqx_recon"
 "emqx_retainer"
-# "emqx_rule_engine" # moved to lib-ce
+"emqx_rule_engine"
 "emqx_sasl"
 "emqx_sn"
 "emqx_stomp"