feat(topic): add few more topic-space-related facilities

apps/emqx/src/emqx_topic.erl

@@ -34,7 +34,9 @@
     systop/1,
     parse/1,
     parse/2,
-    intersection/2
+    intersection/2,
+    subset/2,
+    union/1
 ]).
 
 -export([
@@ -108,10 +110,14 @@ match(_, _) ->
 %% The intersection of two filters is either false or a new topic filter that would match only those topics,
 %% that can be matched by both input filters.
 %% For example, the intersection of "t/global/#" and "t/+/1/+" is "t/global/1/+".
--spec intersection(TopicOrFilter, TopicOrFilter) -> TopicOrFilter | false when
-    TopicOrFilter :: emqx_types:topic().
-intersection(Topic1, Topic2) when is_binary(Topic1), is_binary(Topic2) ->
-    case intersect_start(words(Topic1), words(Topic2)) of
+-spec intersection(TopicOrFilter, TopicOrFilter) -> topic() | false when
+    TopicOrFilter :: topic() | words().
+intersection(Topic1, Topic2) when is_binary(Topic1) ->
+    intersection(words(Topic1), Topic2);
+intersection(Topic1, Topic2) when is_binary(Topic2) ->
+    intersection(Topic1, words(Topic2));
+intersection(Topic1, Topic2) ->
+    case intersect_start(Topic1, Topic2) of
         false -> false;
         Intersection -> join(Intersection)
     end.
@@ -150,6 +156,29 @@ intersect_join(W, Words) -> [W | Words].
 wildcard_intersection(W, W) -> W;
 wildcard_intersection(_, _) -> '+'.
 
+%% @doc Finds out if topic / topic filter T1 is a subset of topic / topic filter T2.
+-spec subset(topic() | words(), topic() | words()) -> boolean().
+subset(T1, T1) ->
+    true;
+subset(T1, T2) when is_binary(T1) ->
+    intersection(T1, T2) =:= T1;
+subset(T1, T2) ->
+    intersection(T1, T2) =:= join(T1).
+
+%% @doc Compute the smallest set of topics / topic filters that contain (have as a
+%% subset) each given topic / topic filter.
+%% Resulting set is not optimal, i.e. it's still possible to have a pair of topic
+%% filters with non-empty intersection.
+-spec union(_Set :: [topic() | words()]) -> [topic() | words()].
+union([Filter | Filters]) ->
+    %% Drop filters completely covered by `Filter`.
+    Disjoint = [F || F <- Filters, not subset(F, Filter)],
+    %% Drop `Filter` if completely covered by another filter.
+    Head = [Filter || not lists:any(fun(F) -> subset(Filter, F) end, Disjoint)],
+    Head ++ union(Disjoint);
+union([]) ->
+    [].
+
 -spec match_share(Name, Filter) -> boolean() when
     Name :: share(),
     Filter :: topic() | share().

apps/emqx/test/emqx_topic_SUITE.erl

@@ -186,6 +186,32 @@ t_sys_intersect(_) ->
     false = intersection(<<"$SYS/broker">>, <<"+/+">>),
     false = intersection(<<"$SYS/broker">>, <<"#">>).
 
+t_union(_) ->
+    Union = fun(Topics) -> lists:sort(emqx_topic:union(Topics)) end,
+    ?assertEqual([], Union([])),
+    ?assertEqual([<<"t/1/2/+">>], Union([<<"t/1/2/+">>])),
+    ?assertEqual([<<"t/1/2/#">>], Union([<<"t/1/2/+">>, <<"t/1/2/#">>])),
+    ?assertEqual([<<"t/+/1/#">>, <<"t/1/+/#">>], Union([<<"t/+/1/#">>, <<"t/1/+/#">>])),
+    ?assertEqual(
+        [<<"g/1">>, <<"t/+/+/#">>, <<"t/1">>],
+        Union([<<"t/1">>, <<"t/1">>, <<"t/2/+">>, <<"t/+/+/#">>, <<"t/+/+">>, <<"g/1">>])
+    ),
+    ?assertEqual(
+        [<<"#">>],
+        Union([<<"t/1">>, <<"t/1">>, <<"t/2/+">>, <<"t/+/+/#">>, <<"#">>, <<"g/1">>])
+    ).
+
+t_union_sys(_) ->
+    Union = fun(Topics) -> lists:sort(emqx_topic:union(Topics)) end,
+    ?assertEqual(
+        [<<"$SYS/cluster">>, <<"+/#">>],
+        Union([<<"+/#">>, <<"t/1/2/#">>, <<"$SYS/cluster">>])
+    ),
+    ?assertEqual(
+        [<<"$SYS/+">>, <<"$SYS/cluster/#">>, <<"+/#">>],
+        Union([<<"+/#">>, <<"t/1/2">>, <<"$SYS/cluster">>, <<"$SYS/cluster/#">>, <<"$SYS/+">>])
+    ).
+
 t_validate(_) ->
     true = validate(<<"a/+/#">>),
     true = validate(<<"a/b/c/d">>),