From 4b3b29873a390849b62438a365619208052928c1 Mon Sep 17 00:00:00 2001
From: JianBo He <heeejianbo@gmail.com>
Date: Fri, 3 Dec 2021 14:41:42 +0800
Subject: [PATCH] feat(retainer): allow to stop publish clear msg

---
 apps/emqx_retainer/etc/emqx_retainer.conf       |  7 +++++++
 apps/emqx_retainer/src/emqx_retainer.erl        | 10 +++++++++-
 apps/emqx_retainer/src/emqx_retainer_schema.erl |  1 +
 3 files changed, 17 insertions(+), 1 deletion(-)

diff --git a/apps/emqx_retainer/etc/emqx_retainer.conf b/apps/emqx_retainer/etc/emqx_retainer.conf
index ba6bdfa6c..92dc62f24 100644
--- a/apps/emqx_retainer/etc/emqx_retainer.conf
+++ b/apps/emqx_retainer/etc/emqx_retainer.conf
@@ -29,6 +29,13 @@ emqx_retainer {
   ## Default: 0s
   msg_expiry_interval = 0s
 
+  ## When the retained flag of the PUBLISH message is set and Payload is empty,
+  ## whether to continue to publish the message.
+  ## see: http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html#_Toc398718038
+  ##
+  ## Default: false
+  #stop_publish_clear_msg = false
+
   ## The message read and deliver flow rate control
   ## When a client subscribe to a wildcard topic, may many retained messages will be loaded.
   ## If you don't want these data loaded to the memory all at once, you can use this to control.
diff --git a/apps/emqx_retainer/src/emqx_retainer.erl b/apps/emqx_retainer/src/emqx_retainer.erl
index 5d248e638..9be449b60 100644
--- a/apps/emqx_retainer/src/emqx_retainer.erl
+++ b/apps/emqx_retainer/src/emqx_retainer.erl
@@ -88,7 +88,12 @@ on_message_publish(Msg = #message{flags   = #{retain := true},
                                   payload = <<>>},
                    Context) ->
     delete_message(Context, Topic),
-    {ok, Msg};
+    case get_stop_publish_clear_msg() of
+        true ->
+            {ok, emqx_message:set_header(allow_publish, false, Msg)};
+        _ ->
+            {ok, Msg}
+    end;
 
 on_message_publish(Msg = #message{flags = #{retain := true}}, Context) ->
     Msg1 = emqx_message:set_header(retained, true, Msg),
@@ -157,6 +162,9 @@ get_expiry_time(#message{timestamp = Ts}) ->
         _ -> Ts + Interval
     end.
 
+get_stop_publish_clear_msg() ->
+    emqx_conf:get([?APP, stop_publish_clear_msg], false).
+
 -spec update_config(hocon:config()) -> ok.
 update_config(Conf) ->
     gen_server:call(?MODULE, {?FUNCTION_NAME, Conf}).
diff --git a/apps/emqx_retainer/src/emqx_retainer_schema.erl b/apps/emqx_retainer/src/emqx_retainer_schema.erl
index 55cfa2fcc..e1fa8373a 100644
--- a/apps/emqx_retainer/src/emqx_retainer_schema.erl
+++ b/apps/emqx_retainer/src/emqx_retainer_schema.erl
@@ -14,6 +14,7 @@ fields("emqx_retainer") ->
     , {msg_clear_interval, sc(emqx_schema:duration_ms(), "0s")}
     , {flow_control, ?TYPE(hoconsc:ref(?MODULE, flow_control))}
     , {max_payload_size, sc(emqx_schema:bytesize(), "1MB")}
+    , {stop_publish_clear_msg, sc(boolean(), false)}
     , {config, config()}
     ];