refactor: apply review remarks

apps/emqx_message_transformation/src/emqx_message_transformation.erl

@@ -112,27 +112,12 @@ unregister_hooks() ->
 
 -spec on_message_publish(emqx_types:message()) ->
     {ok, emqx_types:message()} | {stop, emqx_types:message()}.
-on_message_publish(Message = #message{topic = Topic, headers = Headers}) ->
+on_message_publish(Message = #message{topic = Topic}) ->
     case emqx_message_transformation_registry:matching_transformations(Topic) of
         [] ->
             ok;
         Transformations ->
-            case run_transformations(Transformations, Message) of
+            run_transformations(Transformations, Message)
-                #message{} = FinalMessage ->
-                    emqx_metrics:inc('messages.transformation_succeeded'),
-                    {ok, FinalMessage};
-                drop ->
-                    emqx_metrics:inc('messages.transformation_failed'),
-                    {stop, Message#message{headers = Headers#{allow_publish => false}}};
-                disconnect ->
-                    emqx_metrics:inc('messages.transformation_failed'),
-                    {stop, Message#message{
-                        headers = Headers#{
-                            allow_publish => false,
-                            should_disconnect => true
-                        }
-                    }}
-            end
     end.
 
 %%------------------------------------------------------------------------------
@@ -224,7 +209,25 @@ map_result(RetainBin, [<<"retain">>]) ->
 map_result(Rendered, _Key) ->
     {ok, Rendered}.
 
-run_transformations(Transformations, Message) ->
+run_transformations(Transformations, Message = #message{headers = Headers}) ->
+    case do_run_transformations(Transformations, Message) of
+        #message{} = FinalMessage ->
+            emqx_metrics:inc('messages.transformation_succeeded'),
+            {ok, FinalMessage};
+        drop ->
+            emqx_metrics:inc('messages.transformation_failed'),
+            {stop, Message#message{headers = Headers#{allow_publish => false}}};
+        disconnect ->
+            emqx_metrics:inc('messages.transformation_failed'),
+            {stop, Message#message{
+                headers = Headers#{
+                    allow_publish => false,
+                    should_disconnect => true
+                }
+            }}
+    end.
+
+do_run_transformations(Transformations, Message) ->
     Fun = fun(Transformation, MessageAcc) ->
         #{name := Name} = Transformation,
         emqx_message_transformation_registry:inc_matched(Name),

apps/emqx_message_transformation/src/emqx_message_transformation_registry.erl

@@ -94,18 +94,34 @@ delete(Transformation) ->
 %% @doc Returns a list of matching transformation names, sorted by their configuration order.
 -spec matching_transformations(emqx_types:topic()) -> [transformation()].
 matching_transformations(Topic) ->
-    Transformations0 = [
+    Transformations0 =
-        {Pos, Transformation}
+        lists:flatmap(
-     || M <- emqx_topic_index:matches(Topic, ?TRANSFORMATION_TOPIC_INDEX, [unique]),
+            fun(M) ->
-        [Pos] <- [emqx_topic_index:get_record(M, ?TRANSFORMATION_TOPIC_INDEX)],
+                case emqx_topic_index:get_record(M, ?TRANSFORMATION_TOPIC_INDEX) of
-        {ok, Transformation} <- [
+                    [Pos] ->
-            lookup(emqx_topic_index:get_id(M))
+                        [{Pos, emqx_topic_index:get_id(M)}];
-        ]
+                    _ ->
-    ],
+                        []
-    Transformations1 = lists:sort(
+                end
-        fun({Pos1, _V1}, {Pos2, _V2}) -> Pos1 =< Pos2 end, Transformations0
+            end,
+            emqx_topic_index:matches(Topic, ?TRANSFORMATION_TOPIC_INDEX, [unique])
+        ),
+    Transformations1 =
+        lists:flatmap(
+            fun({Pos, Id}) ->
+                case lookup(Id) of
+                    {ok, Transformation} ->
+                        [{Pos, Transformation}];
+                    _ ->
+                        []
+                end
+            end,
+            Transformations0
+        ),
+    Transformations2 = lists:sort(
+        fun({Pos1, _V1}, {Pos2, _V2}) -> Pos1 =< Pos2 end, Transformations1
     ),
-    lists:map(fun({_Pos, V}) -> V end, Transformations1).
+    lists:map(fun({_Pos, V}) -> V end, Transformations2).
 
 -spec metrics_worker_spec() -> supervisor:child_spec().
 metrics_worker_spec() ->