feat: add topic filter setting

Author: savonarola

priv/config.hocon

@@ -9,6 +9,7 @@ mysql {
         keyfile = "/certs/mysql-client.key"
     }
     server = "mysql:3306"
+    topics = []
     pool_size = 8
     username = "emqx"
     password = "public"

priv/config_i18n.json

@@ -7,6 +7,14 @@
         "en": "Enable MySQL integration",
         "zh": "启用 MySQL 集成"
     },
+    "$mysql_topics_label": {
+        "en": "Topics to persist",
+        "zh": "要持久化的主题"
+    },
+    "$mysql_topics_desc": {
+        "en": "Topic filters for topics to persist, separated by commas",
+        "zh": "要持久化的主题过滤器，用逗号分隔"
+    },
     "$mysql_insert_message_sql_label": {
         "en": "MySQL Insert Message SQL",
         "zh": "MySQL 插入消息 SQL"
@@ -209,6 +217,14 @@
         "en": "Redis Database",
         "zh": "Redis 数据库"
     },
+    "$redis_topics_label": {
+        "en": "Topics to persist",
+        "zh": "要持久化的主题"
+    },
+    "$redis_topics_desc": {
+        "en": "Topic filters for topics to persist, separated by commas",
+        "zh": "要持久化的主题过滤器，用逗号分隔"
+    },
     "$redis_pool_size_label": {
         "en": "Redis Pool Size",
         "zh": "Redis 连接池大小"

priv/config_schema.avsc

@@ -129,6 +129,20 @@
               "description": "$mysql_password_desc"
             }
           },
+          {
+            "name": "topics",
+            "type": {
+              "type": "array",
+              "items": "string"
+            },
+            "default": [],
+            "$ui": {
+              "component": "input-array",
+              "required": false,
+              "label": "$mysql_topics_label",
+              "description": "$mysql_topics_desc"
+            }
+          },
           {
             "name": "batch_size",
             "type": "int",
@@ -309,6 +323,20 @@
               "description": "$redis_database_desc"
             }
           },
+          {
+            "name": "topics",
+            "type": {
+              "type": "array",
+              "items": "string"
+            },
+            "default": [],
+            "$ui": {
+              "component": "input-array",
+              "required": false,
+              "label": "$redis_topics_label",
+              "description": "$redis_topics_desc"
+            }
+          },
           {
             "name": "pool_size",
             "type": "int",

src/emqx_omp.erl

@@ -63,4 +63,3 @@ on_config_changed(OldConf, NewConf) ->
     NewRedisConf = maps:get(<<"redis">>, NewConf, DefaultConf),
     ok = emqx_omp_redis:on_config_changed(OldRedisConf, NewRedisConf),
     ok.
-

src/emqx_omp_mysql.erl

@@ -70,7 +70,11 @@ start(ConfigRaw) ->
         [delete_message_sql, select_message_sql, insert_subscription_sql, select_subscriptions_sql],
         ConfigRaw
     ),
-    Context = Statements,
+    TopicFilters = emqx_omp_utils:topic_filters(ConfigRaw),
+    Context = #{
+        statements => Statements,
+        topic_filters => TopicFilters
+    },
     hook(Context).
 
 %%-------------------------------------------------------------------
@@ -80,7 +84,7 @@ start(ConfigRaw) ->
 on_client_connected(
     ClientInfo = #{clientid := ClientId},
     ConnInfo,
-    #{select_subscriptions_sql := {Sql, ParamTemplate}} = _Context
+    #{statements := #{select_subscriptions_sql := {Sql, ParamTemplate}}} = _Context
 ) ->
     ?SLOG(info, #{
         msg => omp_mysql_client_connected,
@@ -119,7 +123,10 @@ on_session_subscribed(
     ok = fetch_and_deliver_messages(ClientId, Topic, Context).
 
 insert_subscription(
-    ClientId, Topic, SubOpts, #{insert_subscription_sql := {Sql, ParamTemplate}} = _Context
+    ClientId,
+    Topic,
+    SubOpts,
+    #{statements := #{insert_subscription_sql := {Sql, ParamTemplate}}} = _Context
 ) ->
     Qos = maps:get(qos, SubOpts, 0),
     Params = render_row(ParamTemplate, #{clientid => ClientId, topic => Topic, qos => Qos}),
@@ -136,7 +143,7 @@ insert_subscription(
     ok.
 
 fetch_and_deliver_messages(
-    ClientId, Topic, #{select_message_sql := {Sql, ParamTemplate}} = _Context
+    ClientId, Topic, #{statements := #{select_message_sql := {Sql, ParamTemplate}}} = _Context
 ) ->
     Params = render_row(ParamTemplate, #{clientid => ClientId, topic => Topic}),
     _ =
@@ -154,7 +161,7 @@ fetch_and_deliver_messages(
         end,
     ok.
 
-on_session_unsubscribed(#{clientid := ClientId}, Topic, Opts, _Env) ->
+on_session_unsubscribed(#{clientid := ClientId}, Topic, Opts, _Context) ->
     ?SLOG(info, #{
         msg => omp_mysql_session_unsubscribed,
         clientid => ClientId,
@@ -163,17 +170,17 @@ on_session_unsubscribed(#{clientid := ClientId}, Topic, Opts, _Env) ->
     }),
     ok.
 
-on_message_publish(Message = #message{topic = <<"$SYS/", _/binary>>}, _Env) ->
+on_message_publish(Message = #message{topic = <<"$SYS/", _/binary>>}, _Context) ->
     {ok, Message};
-on_message_publish(Message, _Context) ->
+on_message_publish(Message, #{topic_filters := TopicFilters} = _Context) ->
     _ =
-        case emqx_message:qos(Message) of
-            0 ->
+        case emqx_omp_utils:need_persist_message(Message, TopicFilters) of
+            false ->
                 ?SLOG(debug, #{
-                    msg => omp_mysql_message_publish_qos0,
+                    msg => omp_mysql_message_publish_skipped,
                     message => Message
                 });
-            _ ->
+            true ->
                 MessageMap = message_to_map(Message),
                 Res = emqx_resource:query(?RESOURCE_ID, {insert_message, MessageMap}),
                 ?SLOG(info, #{
@@ -187,9 +194,7 @@ on_message_publish(Message, _Context) ->
 on_message_acked(
     _ClientInfo = #{clientid := ClientId},
     #message{id = MsgId} = Message,
-    #{
-        delete_message_sql := {Sql, ParamTemplate}
-    }
+    #{statements := #{delete_message_sql := {Sql, ParamTemplate}}} = _Context
 ) ->
     ?SLOG(info, #{
         msg => omp_mysql_message_puback,
@@ -377,7 +382,6 @@ make_mysql_resource_config(#{<<"insert_message_sql">> := InsertMessageStatement}
 
     {MysqlConfig, ResourceOpts}.
 
-
 sync_query(Sql, Params) ->
     emqx_resource:simple_sync_query(?RESOURCE_ID, {sql, Sql, Params, ?TIMEOUT}).
 

src/emqx_omp_redis.erl

@@ -71,7 +71,8 @@ start(ConfigRaw) ->
         ),
         message_ttl => maps:get(
             <<"message_ttl">>, ConfigRaw, ?DEFAULT_MESSAGE_TTL
-        )
+        ),
+        topic_filters => emqx_omp_utils:topic_filters(ConfigRaw)
     },
     hook(Context).
 
@@ -229,15 +230,15 @@ on_session_unsubscribed(#{clientid := ClientId}, Topic, Opts, _Env) ->
 
 on_message_publish(Message = #message{topic = <<"$SYS/", _/binary>>}, _Env) ->
     {ok, Message};
-on_message_publish(Message, #{message_ttl := TTL} = Context) ->
+on_message_publish(Message, #{message_ttl := TTL, topic_filters := TopicFilters} = Context) ->
     _ =
-        case emqx_message:qos(Message) of
-            0 ->
+        case emqx_omp_utils:need_persist_message(Message, TopicFilters) of
+            false ->
                 ?SLOG(debug, #{
                     msg => omp_redis_message_publish_qos0,
                     message => Message
                 });
-            _ ->
+            true ->
                 Topic = emqx_message:topic(Message),
                 MsgId = emqx_message:id(Message),
                 Now = erlang:system_time(millisecond),

src/emqx_omp_redis_connector.erl

@@ -39,9 +39,10 @@ on_batch_query(InstId, Batch, State) ->
                 {cmd, Cmd} -> [Cmd];
                 {cmds, Cmds} -> Cmds
             end
-        end, Batch),
+        end,
+        Batch
+    ),
     emqx_redis:on_query(InstId, {cmds, Cmds}, State).
 
 on_get_status(InstId, State) ->
     emqx_redis:on_get_status(InstId, State).
-

src/emqx_omp_utils.erl

@@ -4,12 +4,16 @@
 
 -module(emqx_omp_utils).
 
+-include_lib("emqx/include/logger.hrl").
+
 -export([
     fix_ssl_config/1,
     make_resource_opts/1,
     check_config/2,
     deliver_messages/2,
-    induce_subscriptions/1
+    induce_subscriptions/1,
+    need_persist_message/2,
+    topic_filters/1
 ]).
 
 fix_ssl_config(#{<<"ssl">> := SslConfig0} = RawConfig) ->
@@ -66,3 +70,23 @@ induce_subscriptions([]) ->
 induce_subscriptions(Subscriptions) ->
     erlang:send(self(), {subscribe, Subscriptions}),
     ok.
+
+topic_filters(ConfigRaw) ->
+    TopicFiltersRaw = maps:get(<<"topics">>, ConfigRaw, []),
+    [emqx_topic:words(TopicFilterRaw) || TopicFilterRaw <- TopicFiltersRaw].
+
+need_persist_message(Message, TopicFilters) ->
+    ?SLOG(debug, #{
+        msg => omp_utils_need_persist_message,
+        message => emqx_message:to_map(Message),
+        topic_filters => TopicFilters,
+        topic => emqx_message:topic(Message)
+    }),
+    is_message_qos_nonzero(Message) andalso does_message_topic_match(Message, TopicFilters).
+
+is_message_qos_nonzero(Message) ->
+    emqx_message:qos(Message) =/= 0.
+
+does_message_topic_match(Message, TopicFilters) ->
+    Topic = emqx_message:topic(Message),
+    lists:any(fun(Filter) -> emqx_topic:match(Topic, Filter) end, TopicFilters).

test/emqx_omp_SUITE.erl

@@ -47,7 +47,6 @@ init_per_suite(Config) ->
     ok = emqx_omp_test_api_helpers:upload_plugin(Filename),
     ok = emqx_omp_test_api_helpers:start_plugin(PluginId),
     PluginConfig = plugin_config(),
-
     [{plugin_id, PluginId}, {plugin_filename, Filename}, {plugin_config, PluginConfig} | Config].
 
 end_per_suite(_Config) ->
@@ -237,6 +236,7 @@ plugin_config() ->
             enable => false,
             ssl => DefaultSSLConfig#{server_name_indication => <<"redis-server">>},
             servers => <<"invalid-host:6379">>,
+            topics => [<<"t/#">>],
             redis_type => <<"single">>,
             pool_size => 8,
             username => <<"">>,
@@ -252,6 +252,7 @@ plugin_config() ->
         mysql => #{
             enable => false,
             ssl => DefaultSSLConfig#{server_name_indication => <<"mysql-server">>},
+            topics => [<<"t/#">>],
             server => <<"invalid-host:3306">>,
             password => <<"public">>,
             username => <<"emqx">>,

test/emqx_omp_test_api_helpers.erl

@@ -219,4 +219,3 @@ delete_all_actions() ->
 
 asset_path() ->
     filename:join([code:lib_dir(emqx_offline_message_plugin, test), "assets"]).
-