fix(lua_hooks): fix on_message_publish lua hook

CHANGES-4.3.md

@@ -18,6 +18,8 @@ File format:
   management listener. [#8411]
 
 - Fixed crash when shared persistent subscription [#8441]
+- Fixed issue in Lua hook that prevented messages from being
+  rejected [#8535]
 
 ### Enhancements
 - HTTP API(GET /rules/) support for pagination and fuzzy filtering. [#8450]

apps/emqx_lua_hook/src/emqx_lua_hook.app.src

@@ -1,6 +1,6 @@
 {application, emqx_lua_hook,
  [{description, "EMQ X Lua Hooks"},
-  {vsn, "4.3.1"}, % strict semver, bump manually!
+  {vsn, "4.3.2"}, % strict semver, bump manually!
   {modules, []},
   {registered, []},
   {applications, [kernel,stdlib]},

apps/emqx_lua_hook/src/emqx_lua_hook.appup.src

@@ -1,5 +1,7 @@
 %% -*- mode: erlang -*-
 %% Unless you know what you are doing, DO NOT edit manually!!
 {VSN,
-  [{"4.3.0",[{load_module,emqx_lua_script,brutal_purge,soft_purge,[]}]}],
+  [{"4.3.1",[{load_module,emqx_lua_script,brutal_purge,soft_purge,[]}]},
-  [{"4.3.0",[{load_module,emqx_lua_script,brutal_purge,soft_purge,[]}]}]}.
+   {"4.3.0",[{load_module,emqx_lua_script,brutal_purge,soft_purge,[]}]}],
+  [{"4.3.1",[{load_module,emqx_lua_script,brutal_purge,soft_purge,[]}]},
+   {"4.3.0",[{load_module,emqx_lua_script,brutal_purge,soft_purge,[]}]}]}.

apps/emqx_lua_hook/src/emqx_lua_script.erl

@@ -283,13 +283,14 @@ on_message_publish(Message = #message{from = ClientId,
             ?LOG(error, "Failed to execute function on_message_publish(), which has syntax error, St=~p", [St]),
             {ok, Message};
         {[false], _St} ->
-            {stop, Message};
+            ?LOG(debug, "Lua function on_message_publish() returned false, setting allow_publish header to false", []),
+            {stop, Message#message{headers = Headers#{allow_publish => false}}};
         {[NewTopic, NewPayload, NewQos, NewRetain], _St} ->
-            ?LOG(debug, "Lua function on_message_publish() return ~p", [{NewTopic, NewPayload, NewQos, NewRetain}]),
+            ?LOG(debug, "Lua function on_message_publish() returned ~p", [{NewTopic, NewPayload, NewQos, NewRetain}]),
             {ok, Message#message{topic = NewTopic, payload = NewPayload,
                                  qos = round(NewQos), flags = Flags#{retain => to_retain(NewRetain)}}};
         Other ->
-            ?LOG(error, "Topic=~p, lua function on_message_publish caught exception, ~p", [Topic, Other]),
+            ?LOG(error, "Topic=~p, lua function on_message_publish() caught exception, ~p", [Topic, Other]),
             {ok, Message}
     end.
 

apps/emqx_lua_hook/test/emqx_lua_hook_SUITE.erl

@@ -92,7 +92,7 @@ case02(_Config) ->
 
     Msg = #message{from = <<"myclient">>, qos = 2, flags = #{retain => true}, topic = <<"a/b/c">>, payload = <<"123">>, headers = #{username => <<"tester">>}},
     Ret = emqx_hooks:run_fold('message.publish',[], Msg),
-    ?assertEqual(Msg, Ret).
+    ?assertEqual(Msg#message{headers = #{username => <<"tester">>, allow_publish => false}}, Ret).
 
 case03(_Config) ->
     ScriptName = filename:join([emqx_lua_hook:lua_dir(), "abc.lua"]),

src/emqx.appup.src

@@ -2,7 +2,8 @@
 %% Unless you know what you are doing, DO NOT edit manually!!
 {VSN,
   [{"4.3.17",
-    [{load_module,emqx_session,brutal_purge,soft_purge,[]},
+    [{load_module,emqx_exclusive_subscription,brutal_purge,soft_purge,[]},
+     {load_module,emqx_session,brutal_purge,soft_purge,[]},
      {load_module,emqx_shared_sub,brutal_purge,soft_purge,[]},
      {update,emqx_broker_sup,supervisor},
      {load_module,emqx_app,brutal_purge,soft_purge,[]},
@@ -677,7 +678,8 @@
      {load_module,emqx_limiter,brutal_purge,soft_purge,[]}]},
    {<<".*">>,[]}],
   [{"4.3.17",
-    [{load_module,emqx_session,brutal_purge,soft_purge,[]},
+    [{load_module,emqx_exclusive_subscription,brutal_purge,soft_purge,[]},
+     {load_module,emqx_session,brutal_purge,soft_purge,[]},
      {load_module,emqx_shared_sub,brutal_purge,soft_purge,[]},
      {update,emqx_broker_sup,supervisor},
      {load_module,emqx_app,brutal_purge,soft_purge,[]},