feat(redis): add a rule function to help formatting redis args

The new function named 'map_to_redis_hset_args' can be used to format
a map's key-value pairs into redis HSET (or HMSET) arg list.

This new function is dedicated for redis to avoid abuse for other
data integrations.

apps/emqx_bridge_redis/src/emqx_bridge_redis_connector.erl

@@ -128,8 +128,8 @@ on_query(
                 #{instance_id => InstId, cmd => Cmd, batch => false, mode => sync, result => Result}
             ),
             Result;
-        Error ->
-            Error
+        {error, Reason} ->
+            {error, Reason}
     end.
 
 on_batch_query(
@@ -165,8 +165,8 @@ on_batch_query(
                 }
             ),
             Result;
-        Error ->
-            Error
+        {error, Reason} ->
+            {error, Reason}
     end.
 
 trace_format_commands(Commands0) ->
@@ -204,11 +204,15 @@ query(InstId, Query, RedisConnSt) ->
     end.
 
 proc_command_template(CommandTemplate, Msg) ->
-    lists:map(
-        fun(ArgTks) ->
-            emqx_placeholder:proc_tmpl(ArgTks, Msg, #{return => full_binary})
-        end,
-        CommandTemplate
+    lists:reverse(
+        lists:foldl(
+            fun(ArgTks, Acc) ->
+                New = proc_tmpl(ArgTks, Msg),
+                lists:reverse(New, Acc)
+            end,
+            [],
+            CommandTemplate
+        )
     ).
 
 preproc_command_template(CommandTemplate) ->
@@ -216,3 +220,18 @@ preproc_command_template(CommandTemplate) ->
         fun emqx_placeholder:preproc_tmpl/1,
         CommandTemplate
     ).
+
+%% This function mimics emqx_placeholder:proc_tmpl/3 but with an
+%% injected special handling of map_to_redis_hset_args result
+%% which is a list of redis command args (all in binary string format)
+proc_tmpl([{var, Phld}], Data) ->
+    case emqx_placeholder:lookup_var(Phld, Data) of
+        [map_to_redis_hset_args | L] ->
+            L;
+        Other ->
+            [emqx_utils_conv:bin(Other)]
+    end;
+proc_tmpl(Tokens, Data) ->
+    %% more than just a var ref, but a string, or a concatenation of string and a var
+    %% this is must be a single arg, format it into a binary
+    [emqx_placeholder:proc_tmpl(Tokens, Data, #{return => full_binary})].

apps/emqx_rule_engine/src/emqx_rule_funcs.erl

@@ -160,6 +160,7 @@
     find/3,
     join_to_string/1,
     join_to_string/2,
+    map_to_redis_hset_args/1,
     join_to_sql_values_string/1,
     jq/2,
     jq/3,
@@ -814,6 +815,38 @@ join_to_string(Str) -> emqx_variform_bif:join_to_string(Str).
 
 join_to_string(Sep, List) -> emqx_variform_bif:join_to_string(Sep, List).
 
+%% @doc Format map key-value pairs as redis HSET (or HMSET) command fields.
+%% Notes:
+%% - Non-string keys in the input map are dropped
+%% - Keys are not quoted
+%% - String values are always quoted
+%% - No escape sequence for keys and values
+%% - Float point values are formatted with fixed (6) decimal point compact-formatting
+map_to_redis_hset_args(Map) when erlang:is_map(Map) ->
+    [map_to_redis_hset_args | maps:fold(fun redis_hset_acc/3, [], Map)].
+
+redis_hset_acc(K, V, IoData) ->
+    try
+        [redis_field_name(K), redis_field_value(V) | IoData]
+    catch
+        _:_ ->
+            IoData
+    end.
+
+redis_field_name(K) when erlang:is_binary(K) ->
+    K;
+redis_field_name(K) ->
+    throw({bad_redis_field_name, K}).
+
+redis_field_value(V) when erlang:is_binary(V) ->
+    iolist_to_binary([$", V, $"]);
+redis_field_value(V) when erlang:is_integer(V) ->
+    integer_to_binary(V);
+redis_field_value(V) when erlang:is_float(V) ->
+    float2str(V, 6);
+redis_field_value(V) when erlang:is_boolean(V) ->
+    atom_to_binary(V).
+
 join_to_sql_values_string(List) ->
     QuotedList =
         [

apps/emqx_rule_engine/test/emqx_rule_funcs_SUITE.erl

@@ -1376,6 +1376,27 @@ t_parse_date_errors(_) ->
 
     ok.
 
+t_map_to_redis_hset_args(_Config) ->
+    Do = fun(Map) -> tl(emqx_rule_funcs:map_to_redis_hset_args(Map)) end,
+    ?assertEqual([], Do(#{})),
+    ?assertEqual([], Do(#{1 => 2})),
+    ?assertEqual([<<"a">>, <<"1">>], Do(#{<<"a">> => 1, 3 => 4})),
+    ?assertEqual([<<"a">>, <<"1.1">>], Do(#{<<"a">> => 1.1})),
+    ?assertEqual([<<"a">>, <<"true">>], Do(#{<<"a">> => true})),
+    ?assertEqual([<<"a">>, <<"false">>], Do(#{<<"a">> => false})),
+    ?assertEqual([<<"a">>, <<"\"\"">>], Do(#{<<"a">> => <<"">>})),
+    ?assertEqual([<<"a">>, <<"\"i j\"">>], Do(#{<<"a">> => <<"i j">>})),
+    %% no determined ordering
+    ?assert(
+        case Do(#{<<"a">> => 1, <<"b">> => 2}) of
+            [<<"a">>, <<"1">>, <<"b">>, <<"2">>] ->
+                true;
+            [<<"b">>, <<"2">>, <<"a">>, <<"1">>] ->
+                true
+        end
+    ),
+    ok.
+
 %%------------------------------------------------------------------------------
 %% Utility functions
 %%------------------------------------------------------------------------------

apps/emqx_utils/src/emqx_placeholder.erl

@@ -37,7 +37,8 @@
     proc_tmpl_deep/3,
 
     bin/1,
-    sql_data/1
+    sql_data/1,
+    lookup_var/2
 ]).
 
 -export([

apps/emqx_utils/src/emqx_utils.app.src

@@ -2,7 +2,7 @@
 {application, emqx_utils, [
     {description, "Miscellaneous utilities for EMQX apps"},
     % strict semver, bump manually!
-    {vsn, "5.2.0"},
+    {vsn, "5.2.1"},
     {modules, [
         emqx_utils,
         emqx_utils_api,