Merge pull request #9942 from emqx/http-api-publish-crash

Http api publish crash

apps/emqx_rule_engine/src/emqx_rule_engine.appup.src

@@ -3,11 +3,13 @@
 {VSN,
   [{<<"4\\.4\\.1[3-4]">>,
     [{load_module,emqx_rule_registry,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_events,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_monitor,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine,brutal_purge,soft_purge,[]}
     ]},
    {"4.4.12",
     [{load_module,emqx_rule_registry,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_events,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_monitor,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine_jwt_worker,brutal_purge,soft_purge,[]},
@@ -15,6 +17,7 @@
      {load_module,emqx_rule_engine_jwt,brutal_purge,soft_purge,[]}]},
    {"4.4.11",
     [{load_module,emqx_rule_engine_jwt_worker,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_events,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_monitor,brutal_purge,soft_purge,[]},
      {update,emqx_rule_engine_jwt_sup,supervisor},
      {load_module,emqx_rule_engine_jwt,brutal_purge,soft_purge,[]},
@@ -231,11 +234,13 @@
    {<<".*">>,[]}],
   [{<<"4\\.4\\.1[3-4]">>,
     [{load_module,emqx_rule_registry,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_events,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_monitor,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine,brutal_purge,soft_purge,[]}
     ]},
    {"4.4.12",
     [{load_module,emqx_rule_registry,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_events,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_monitor,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine_jwt_worker,brutal_purge,soft_purge,[]},
@@ -243,6 +248,7 @@
      {load_module,emqx_rule_engine_jwt,brutal_purge,soft_purge,[]}]},
    {"4.4.11",
     [{load_module,emqx_rule_engine_jwt_worker,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_events,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_monitor,brutal_purge,soft_purge,[]},
      {update,emqx_rule_engine_jwt_sup,supervisor},
      {load_module,emqx_rule_engine_jwt,brutal_purge,soft_purge,[]},

apps/emqx_rule_engine/src/emqx_rule_events.erl

@@ -853,9 +853,25 @@ printable_maps(Headers) ->
                         value => Value
                      } || {Key, Value} <- V0]
                 };
+            (K, V, AccIn) when is_map(V) ->
+                AccIn#{K => printable_maps(V)};
+            (K, V, AccIn) when is_list(V) ->
+                AccIn#{K => printable_list(V)};
             (_K, V, AccIn) when is_tuple(V) ->
-                %% internal header
+                %% internal header, remove it
                 AccIn;
             (K, V, AccIn) ->
                 AccIn#{K => V}
         end, #{}, Headers).
+
+printable_list(L) ->
+    lists:filtermap(fun printable_element/1, L).
+
+printable_element(E) when is_map(E) ->
+    {true, printable_maps(E)};
+printable_element(E) when is_tuple(E) ->
+    false;
+printable_element(E) when is_list(E) ->
+    {true, printable_list(E)};
+printable_element(E) ->
+    {true, E}.

apps/emqx_rule_engine/test/emqx_rule_events_SUITE.erl

@@ -36,18 +36,44 @@ t_mod_hook_fun(_) ->
               ]].
 
 t_printable_maps(_) ->
-    Headers = #{peerhost => {127,0,0,1},
+    TestMap = #{
+      peerhost => {127,0,0,1},
       peername => {{127,0,0,1}, 9980},
       sockname => {{127,0,0,1}, 1883},
       redispatch_to => {<<"group">>, <<"sub/topic/+">>},
-                shared_dispatch_ack => {self(), ref}
+      shared_dispatch_ack => {self(), ref},
+      string => <<"abc">>,
+      atom => abc,
+      integer => 1,
+      float => 1.0,
+      simple_list => [1, 1.0, a, "abc", <<"abc">>, {a,b}]
+    },
+    Headers = TestMap#{
+      map => TestMap,
+      map_list => [
+        TestMap#{
+          map => TestMap
+        }
+      ]
     },
     Converted = emqx_rule_events:printable_maps(Headers),
+    Verify = fun(Result) ->
       ?assertMatch(
           #{peerhost := <<"127.0.0.1">>,
             peername := <<"127.0.0.1:9980">>,
-          sockname := <<"127.0.0.1:1883">>
+            sockname := <<"127.0.0.1:1883">>,
-        }, Converted),
+            string := <<"abc">>,
-    ?assertNot(maps:is_key(redispatch_to, Converted)),
+            atom := abc,
-    ?assertNot(maps:is_key(shared_dispatch_ack, Converted)),
+            integer := 1,
+            float := 1.0,
+            simple_list := [1, 1.0, a, "abc", <<"abc">>] %% {a,b} is removed
+          }, Result),
+      ?assertNot(maps:is_key(redispatch_to, Result)),
+      ?assertNot(maps:is_key(shared_dispatch_ack, Result)),
+      %% make sure the result is jsonable
+      _ = emqx_json:encode(Result)
+    end,
+    Verify(maps:get(map, Converted)),
+    Verify(maps:get(map, lists:nth(1, maps:get(map_list, Converted)))),
+    Verify(Converted),
     ok.

apps/emqx_rule_engine/test/emqx_rule_monitor_SUITE.erl

@@ -90,6 +90,7 @@ end_per_testcase(_, Config) ->
 
 common_init_per_testcase() ->
     AlarmOpts = [{actions, [log, publish]}, {size_limit, 1000}, {validity_period, 86400}],
+    _ = emqx_alarm:mnesia(boot),
     {ok, _} = emqx_alarm:start_link(AlarmOpts),
     {ok, _} = emqx_rule_monitor:start_link().
 

changes/v4.4.15-en.md

@@ -16,6 +16,8 @@
 
 ## Bug fixes
 
+- Fixed an error when forward MQTT messages with User-Property using the `republish` action [#9942](https://github.com/emqx/emqx/pull/9942).
+
 - fix some issues in descriptions of the actions, resources amd emqx-modules [#9931](https://github.com/emqx/emqx/pull/9931).
 
 - fix there's no error logs when query the JWKS server failed [#9931](https://github.com/emqx/emqx/pull/9931).

changes/v4.4.15-zh.md

@@ -16,6 +16,8 @@
 
 ## 修复
 
+- 修复使用 `消息重发布` 动作转发带 User-Property 的 MQTT 消息时出错的问题 [#9942](https://github.com/emqx/emqx/pull/9942)。
+
 - 修复资源、动作以及模块里的一些描述错误 [#9931](https://github.com/emqx/emqx/pull/9931)。
 
 - 修复请求 JWKS 服务失败的时候，没有日志打印的问题 [#9931](https://github.com/emqx/emqx/pull/9931)。