MQTT: Support wildcards in topic filters matching retained messages

deps/rabbitmq_ct_client_helpers/MODULE.bazel

@@ -0,0 +1,6 @@
+###############################################################################
+# Bazel now uses Bzlmod by default to manage external dependencies.
+# Please consider migrating your external dependencies from WORKSPACE to MODULE.bazel.
+#
+# For more details, please check https://github.com/bazelbuild/bazel/issues/18958
+###############################################################################

deps/rabbitmq_mqtt/Makefile

@@ -12,6 +12,7 @@ define PROJECT_ENV
 	    {retained_message_store, rabbit_mqtt_retained_msg_store_dets},
 	    %% only used by DETS store
 	    {retained_message_store_dets_sync_interval, 2000},
+			{retained_message_store_max_retained_messages_count, 2000},
 	    {prefetch, 10},
 	    {ssl_listeners, []},
 	    {tcp_listeners, [1883]},
@@ -94,7 +95,7 @@ define ct_master.erl
 	halt(0)
 endef
 
-PARALLEL_CT_SET_1_A = auth retainer federation feature_flag
+PARALLEL_CT_SET_1_A = auth rabbit_mqtt_retained_msg_store retainer federation feature_flag
 PARALLEL_CT_SET_1_B = cluster command config config_schema mc_mqtt packet_prop \
     processor protocol_interop proxy_protocol rabbit_mqtt_confirms reader util
 PARALLEL_CT_SET_1_C = java v5

deps/rabbitmq_mqtt/priv/schema/rabbitmq_mqtt.schema

@@ -87,6 +87,10 @@ end}.
 {mapping, "mqtt.retained_message_store_dets_sync_interval", "rabbitmq_mqtt.retained_message_store_dets_sync_interval",
     [{datatype, integer}]}.
 
+%% Limit how many messages are returned by MQTT plugin retained messages store
+{mapping, "mqtt.retained_message_store_max_retained_messages_count", "rabbitmq_mqtt.retained_message_store_max_retained_messages_count",
+    [{datatype, integer}]}.
+
 %% Whether or not to enable proxy protocol support.
 %%
 %% {proxy_protocol, false}

deps/rabbitmq_mqtt/src/rabbit_mqtt_processor.erl

@@ -998,44 +998,53 @@ send_retained_messages(Subscriptions, State) ->
 
 -spec send_retained_message(topic_filter(), qos(), state()) -> state().
 send_retained_message(TopicFilter0, SubscribeQos,
-                      State0 = #state{packet_id = PacketId0,
-                                      cfg = #cfg{retainer_pid = RPid}}) ->
+                     State0 = #state{cfg = #cfg{retainer_pid = RPid}}) ->
     TopicFilter = amqp_to_mqtt(TopicFilter0),
     case rabbit_mqtt_retainer:fetch(RPid, TopicFilter) of
         undefined ->
             State0;
-        #mqtt_msg{qos = MsgQos,
-                  retain = Retain,
-                  payload = Payload,
-                  props = Props0} ->
-            Qos = effective_qos(MsgQos, SubscribeQos),
-            %% Wildcards are currently not supported when fetching retained
-            %% messages. Therefore, TopicFilter must must be a topic name.
-            {Topic, Props, State1} = process_topic_alias_outbound(TopicFilter, Props0, State0),
-            {PacketId, State} = case Qos of
-                                    ?QOS_0 ->
-                                        {undefined, State1};
-                                    ?QOS_1 ->
-                                        {PacketId0,
-                                         State1#state{packet_id = increment_packet_id(PacketId0)}}
-                                end,
-            Packet = #mqtt_packet{
-                        fixed = #mqtt_packet_fixed{
-                                   type = ?PUBLISH,
-                                   qos  = Qos,
-                                   dup  = false,
-                                   retain = Retain
-                                  },
-                        variable = #mqtt_packet_publish{
-                                      packet_id = PacketId,
-                                      topic_name = Topic,
-                                      props = Props
-                                     },
-                        payload = Payload},
-            _ = send(Packet, State),
-            State
+        Msgs when is_list(Msgs) ->
+            lists:foldl(
+              fun(Msg, S) ->
+                  send_retained_message_to_client(Msg, SubscribeQos, S)
+              end, State0, Msgs);
+        #mqtt_msg{} = SingleMsg ->
+            send_retained_message_to_client(SingleMsg, SubscribeQos, State0)
     end.
 
+send_retained_message_to_client(#mqtt_msg{qos = MsgQos,
+                                        retain = Retain,
+                                        topic = Topic0,
+                                        payload = Payload,
+                                        props = Props0},
+                               SubscribeQos,
+                               State0 = #state{packet_id = PacketId0}) ->
+    Qos = effective_qos(MsgQos, SubscribeQos),
+    Topic1 = amqp_to_mqtt(Topic0),
+    {Topic, Props, State1} = process_topic_alias_outbound(Topic1, Props0, State0),
+    {PacketId, State} = case Qos of
+                           ?QOS_0 ->
+                               {undefined, State1};
+                           ?QOS_1 ->
+                               {PacketId0,
+                                State1#state{packet_id = increment_packet_id(PacketId0)}}
+                       end,
+    Packet = #mqtt_packet{
+               fixed = #mqtt_packet_fixed{
+                        type = ?PUBLISH,
+                        qos  = Qos,
+                        dup  = false,
+                        retain = Retain
+                       },
+               variable = #mqtt_packet_publish{
+                          packet_id = PacketId,
+                          topic_name = Topic,
+                          props = Props
+                         },
+               payload = Payload},
+    _ = send(Packet, State),
+    State.
+
 clear_will_msg(#state{cfg = #cfg{vhost = Vhost,
                                  client_id = ClientId}} = State) ->
     QNameBin = rabbit_mqtt_util:queue_name_bin(ClientId, will),

deps/rabbitmq_mqtt/src/rabbit_mqtt_retained_msg_store.erl

@@ -20,103 +20,97 @@
 
 -include("rabbit_mqtt.hrl").
 -include("rabbit_mqtt_packet.hrl").
+
 -include_lib("kernel/include/logger.hrl").
+
 -export([start/1, insert/3, lookup/2, delete/2, terminate/1]).
--export([expire/2]).
+-export([expire/2, get_max_retained_messages_count/0]).
+
 -export_type([state/0, expire/0]).
 
 -define(STATE, ?MODULE).
--record(?STATE, {store_mod :: module(),
-                 store_state :: term()}).
--opaque state() :: #?STATE{}.
-
--type expire() :: #{topic() :=
-                    {InsertionTimestamp :: integer(),
-                     MessageExpiryInterval :: pos_integer()}}.
 
--callback new(Directory :: file:name_all(), rabbit_types:vhost()) ->
-    State :: any().
+-record(?STATE, {store_mod :: module(), store_state :: term()}).
 
--callback recover(Directory :: file:name_all(), rabbit_types:vhost()) ->
-    {ok, State :: any(), expire()} |
-    {error, uninitialized}.
-
--callback insert(topic(), mqtt_msg(), State :: any()) ->
-    ok.
-
--callback lookup(topic(), State :: any()) ->
-    mqtt_msg() | mqtt_msg_v0() | undefined.
+-opaque state() :: #?STATE{}.
 
--callback delete(topic(), State :: any()) ->
-    ok.
+-type expire() ::
+    #{topic() := {InsertionTimestamp :: integer(), MessageExpiryInterval :: pos_integer()}}.
 
--callback terminate(State :: any()) ->
-    ok.
+-callback new(Directory :: file:name_all(), rabbit_types:vhost()) -> State :: any().
+-callback recover(Directory :: file:name_all(), rabbit_types:vhost()) ->
+                     {ok, State :: any(), expire()} | {error, uninitialized}.
+-callback insert(topic(), mqtt_msg(), State :: any()) -> ok.
+-callback lookup(topic(), State :: any()) -> [mqtt_msg()] | [mqtt_msg_v0()] | [].
+-callback delete(topic(), State :: any()) -> ok.
+-callback terminate(State :: any()) -> ok.
 
 -spec start(rabbit_types:vhost()) -> {state(), expire()}.
 start(VHost) ->
     {ok, Mod} = application:get_env(?APP_NAME, retained_message_store),
     Dir = rabbit:data_dir(),
-    ?LOG_INFO("Starting MQTT retained message store ~s for vhost '~ts'",
-              [Mod, VHost]),
-    {S, Expire} = case Mod:recover(Dir, VHost) of
-                      {ok, StoreState, Expire0} ->
-                          ?LOG_INFO("Recovered MQTT retained message store ~s for vhost '~ts'",
-                                    [Mod, VHost]),
-                          {StoreState, Expire0};
-                      {error, uninitialized} ->
-                          StoreState = Mod:new(Dir, VHost),
-                          ?LOG_INFO("Initialized MQTT retained message store ~s for vhost '~ts'",
-                                    [Mod, VHost]),
-                          {StoreState, #{}}
-                  end,
-    {#?STATE{store_mod = Mod,
-             store_state = S}, Expire}.
+    ?LOG_INFO("Starting MQTT retained message store ~s for vhost '~ts'", [Mod, VHost]),
+    {S, Expire} =
+        case Mod:recover(Dir, VHost) of
+            {ok, StoreState, Expire0} ->
+                ?LOG_INFO("Recovered MQTT retained message store ~s for vhost '~ts'", [Mod, VHost]),
+                {StoreState, Expire0};
+            {error, uninitialized} ->
+                StoreState = Mod:new(Dir, VHost),
+                ?LOG_INFO("Initialized MQTT retained message store ~s for vhost '~ts'",
+                          [Mod, VHost]),
+                {StoreState, #{}}
+        end,
+    {#?STATE{store_mod = Mod, store_state = S}, Expire}.
 
 -spec insert(topic(), mqtt_msg(), state()) -> ok.
-insert(Topic, Msg, #?STATE{store_mod = Mod,
-                           store_state = StoreState}) ->
+insert(Topic, Msg, #?STATE{store_mod = Mod, store_state = StoreState}) ->
     ok = Mod:insert(Topic, Msg, StoreState).
 
--spec lookup(topic(), state()) ->
-    mqtt_msg() | undefined.
-lookup(Topic, #?STATE{store_mod = Mod,
-                      store_state = StoreState}) ->
+-spec lookup(topic(), state()) -> [mqtt_msg()] | [].
+lookup(Topic, #?STATE{store_mod = Mod, store_state = StoreState}) ->
     case Mod:lookup(Topic, StoreState) of
-        OldMsg when is_record(OldMsg, mqtt_msg, 7) ->
-            convert_mqtt_msg(OldMsg);
-        Other ->
-            Other
+        % Handle list of messages - convert any old format ones
+        Messages when is_list(Messages) ->
+            lists:map(fun (Msg) when is_record(Msg, mqtt_msg, 7) ->
+                              convert_mqtt_msg(Msg);
+                          (Msg) ->
+                              Msg
+                      end,
+                      Messages);
+        undefined ->
+            [];
+        [] ->
+            []
     end.
 
 -spec delete(topic(), state()) -> ok.
-delete(Topic, #?STATE{store_mod = Mod,
-                      store_state = StoreState}) ->
+delete(Topic, #?STATE{store_mod = Mod, store_state = StoreState}) ->
     ok = Mod:delete(Topic, StoreState).
 
 -spec terminate(state()) -> ok.
-terminate(#?STATE{store_mod = Mod,
-                  store_state = StoreState}) ->
+terminate(#?STATE{store_mod = Mod, store_state = StoreState}) ->
     ok = Mod:terminate(StoreState).
 
 -spec expire(ets | dets, ets:tid() | dets:tab_name()) -> expire().
 expire(Mod, Tab) ->
     Now = os:system_time(second),
-    Mod:foldl(
-      fun(#retained_message{topic = Topic,
-                            mqtt_msg = #mqtt_msg{props = #{'Message-Expiry-Interval' := Expiry},
-                                                 timestamp = Timestamp}}, Acc)
-            when is_integer(Expiry) andalso
-                 is_integer(Timestamp) ->
-              if Now - Timestamp >= Expiry ->
-                     Mod:delete(Tab, Topic),
-                     Acc;
-                 true ->
-                     maps:put(Topic, {Timestamp, Expiry}, Acc)
-              end;
-         (_, Acc) ->
-              Acc
-      end, #{}, Tab).
+    ExpireMsg =
+        fun ({NodeId,
+              Topic,
+              #mqtt_msg{props = #{'Message-Expiry-Interval' := Expiry}, timestamp = Timestamp}},
+             Acc)
+                when is_integer(Expiry) andalso is_integer(Timestamp) ->
+                if Now - Timestamp >= Expiry ->
+                       Mod:delete(Tab, NodeId),
+                       Acc;
+                   true ->
+                       maps:put(Topic, {Timestamp, Expiry}, Acc)
+                end;
+            (_, Acc) ->
+                Acc
+        end,
+    Mod:foldl(ExpireMsg, #{}, Tab).
 
 %% Retained messages written in 3.12 (or earlier) are converted when read in 3.13 (or later).
 -spec convert_mqtt_msg(mqtt_msg_v0()) -> mqtt_msg().
@@ -128,3 +122,9 @@ convert_mqtt_msg({mqtt_msg, Retain, Qos, Topic, Dup, PacketId, Payload}) ->
               packet_id = PacketId,
               payload = Payload,
               props = #{}}.
+
+-spec get_max_retained_messages_count() -> pos_integer().
+get_max_retained_messages_count() ->
+    rabbit_misc:get_env(rabbit_mqtt,
+                        retained_message_store_max_retained_messages_count,
+                        2000).