fix(rules): parse outputs failed

apps/emqx_rule_engine/src/emqx_rule_engine.erl

@@ -86,7 +86,7 @@ do_create_rule(Params = #{id := RuleId, sql := Sql, outputs := Outputs}) ->
             ok = emqx_rule_registry:add_rule(Rule),
             _ = emqx_plugin_libs_rule:cluster_call(emqx_rule_metrics, create_rule_metrics, [RuleId]),
             {ok, Rule};
-        Reason -> {error, Reason}
+        {error, Reason} -> {error, Reason}
     end.
 
 parse_outputs(Outputs) ->
@@ -97,8 +97,9 @@ do_parse_outputs(#{type := bridge, target := ChId}) ->
 do_parse_outputs(#{type := builtin, target := Repub, args := Args})
         when Repub == republish; Repub == <<"republish">> ->
     #{type => builtin, target => republish, args => pre_process_repub_args(Args)};
-do_parse_outputs(#{type := builtin, target := Name} = Output) ->
-    #{type => builtin, target => Name, args => maps:get(args, Output, #{})}.
+do_parse_outputs(#{type := Type, target := Name} = Output)
+        when Type == func; Type == builtin ->
+    #{type => Type, target => Name, args => maps:get(args, Output, #{})}.
 
 pre_process_repub_args(#{<<"topic">> := Topic} = Args) ->
     QoS = maps:get(<<"qos">>, Args, <<"${qos}">>),

apps/emqx_rule_engine/src/emqx_rule_engine_api.erl

@@ -323,7 +323,7 @@ format_output(Outputs) ->
     [do_format_output(Out) || Out <- Outputs].
 
 do_format_output(#{type := func}) ->
-    #{type => builtin, target => <<"internal_function">>};
+    #{type => func, target => <<"internal_function">>};
 do_format_output(#{type := builtin, target := Name, args := Args}) ->
     #{type => builtin, target => Name, args => maps:remove(preprocessed_tmpl, Args)};
 do_format_output(#{type := bridge, target := Name}) ->

apps/emqx_rule_engine/src/emqx_rule_sqlparser.erl

@@ -53,8 +53,7 @@
 -dialyzer({nowarn_function, [parse/1]}).
 
 %% Parse one select statement.
--spec(parse(string() | binary())
-      -> {ok, select()} | {parse_error, term()} | {lex_error, term()}).
+-spec(parse(string() | binary()) -> {ok, select()} | {error, term()}).
 parse(Sql) ->
     try case rulesql:parsetree(Sql) of
             {ok, {select, Clauses}} ->
@@ -75,11 +74,11 @@ parse(Sql) ->
                         from = get_value(from, Clauses),
                         where = get_value(where, Clauses)
                     }};
-            Error -> Error
+            {error, Error} -> {error, Error}
         end
     catch
         _Error:Reason:StackTrace ->
-            {parse_error, {Reason, StackTrace}}
+            {error, {Reason, StackTrace}}
     end.
 
 -spec(select_fields(select()) -> list(field())).

apps/emqx_rule_engine/test/emqx_rule_engine_SUITE.erl

@@ -203,7 +203,7 @@ t_crud_rule_api(_Config) ->
     {201, Rule} = emqx_rule_engine_api:crud_rules(post, #{body => Params0}),
 
     ?assertEqual(RuleID, maps:get(id, Rule)),
-    {ok, Rules} = emqx_rule_engine_api:crud_rules(get, #{}),
+    {200, Rules} = emqx_rule_engine_api:crud_rules(get, #{}),
     ct:pal("RList : ~p", [Rules]),
     ?assert(length(Rules) > 0),