MQTT: Support wildcards in topic filters matching retained messages

deps/rabbit/BUILD.bazel

@@ -198,17 +198,18 @@ rabbitmq_app(
         "syntax_tools",
         "xmerl",
         "crypto",
+        "horus",
     ],
     license_files = [":license_files"],
     priv = [":priv"],
     deps = [
         "//deps/amqp10_common:erlang_app",
+        "//deps/khepri:erlang_app",
+        "//deps/khepri_mnesia_migration:erlang_app",
         "//deps/rabbit_common:erlang_app",
         "//deps/rabbitmq_prelaunch:erlang_app",
         "@cuttlefish//:erlang_app",
         "@gen_batch_server//:erlang_app",
-        "@khepri//:erlang_app",
-        "@khepri_mnesia_migration//:erlang_app",
         "@observer_cli//:erlang_app",
         "@osiris//:erlang_app",
         "@ra//:erlang_app",

deps/rabbit/app.bzl

@@ -248,9 +248,9 @@ def all_beam_files(name = "all_beam_files"):
         erlc_opts = "//:erlc_opts",
         deps = [
             "//deps/amqp10_common:erlang_app",
+            "//deps/khepri:erlang_app",
+            "//deps/khepri_mnesia_migration:erlang_app",
             "//deps/rabbit_common:erlang_app",
-            "@khepri//:erlang_app",
-            "@khepri_mnesia_migration//:erlang_app",
             "@ra//:erlang_app",
             "@ranch//:erlang_app",
             "@stdout_formatter//:erlang_app",
@@ -507,9 +507,9 @@ def all_test_beam_files(name = "all_test_beam_files"):
         erlc_opts = "//:test_erlc_opts",
         deps = [
             "//deps/amqp10_common:erlang_app",
+            "//deps/khepri:erlang_app",
+            "//deps/khepri_mnesia_migration:erlang_app",
             "//deps/rabbit_common:erlang_app",
-            "@khepri//:erlang_app",
-            "@khepri_mnesia_migration//:erlang_app",
             "@ra//:erlang_app",
             "@ranch//:erlang_app",
             "@stdout_formatter//:erlang_app",
@@ -2021,7 +2021,7 @@ def test_suite_beam_files(name = "test_suite_beam_files"):
         outs = ["test/metadata_store_phase1_SUITE.beam"],
         app_name = "rabbit",
         erlc_opts = "//:test_erlc_opts",
-        deps = ["//deps/rabbit_common:erlang_app", "@khepri//:erlang_app"],
+        deps = ["//deps/khepri:erlang_app", "//deps/rabbit_common:erlang_app"],
     )
     erlang_bytecode(
         name = "mc_unit_SUITE_beam_files",

deps/rabbitmq_consistent_hash_exchange/BUILD.bazel

@@ -39,10 +39,10 @@ rabbitmq_app(
     license_files = [":license_files"],
     priv = [":priv"],
     deps = [
+        "//deps/khepri:erlang_app",
+        "//deps/khepri_mnesia_migration:erlang_app",
         "//deps/rabbit:erlang_app",
         "//deps/rabbit_common:erlang_app",
-        "@khepri//:erlang_app",
-        "@khepri_mnesia_migration//:erlang_app",
     ],
 )
 

deps/rabbitmq_consistent_hash_exchange/app.bzl

@@ -19,11 +19,11 @@ def all_beam_files(name = "all_beam_files"):
         dest = "ebin",
         erlc_opts = "//:erlc_opts",
         deps = [
+            "//deps/khepri:erlang_app",
+            "//deps/khepri_mnesia_migration:erlang_app",
             "//deps/rabbit:erlang_app",
             "//deps/rabbit_common:erlang_app",
             "//deps/rabbitmq_cli:erlang_app",
-            "@khepri//:erlang_app",
-            "@khepri_mnesia_migration//:erlang_app",
         ],
     )
 
@@ -47,11 +47,11 @@ def all_test_beam_files(name = "all_test_beam_files"):
         dest = "test",
         erlc_opts = "//:test_erlc_opts",
         deps = [
+            "//deps/khepri:erlang_app",
+            "//deps/khepri_mnesia_migration:erlang_app",
             "//deps/rabbit:erlang_app",
             "//deps/rabbit_common:erlang_app",
             "//deps/rabbitmq_cli:erlang_app",
-            "@khepri//:erlang_app",
-            "@khepri_mnesia_migration//:erlang_app",
         ],
     )
 

deps/rabbitmq_ct_client_helpers/BUILD.bazel

@@ -33,6 +33,7 @@ rabbitmq_app(
     hdrs = [":public_hdrs"],
     app_name = "rabbitmq_ct_client_helpers",
     beam_files = [":beam_files"],
+    extra_apps = ["rabbit_common"],
     license_files = [":license_files"],
     priv = [":priv"],
     deps = [

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_ct_helpers/BUILD.bazel

@@ -39,6 +39,12 @@ rabbitmq_app(
     hdrs = [":public_hdrs"],
     app_name = "rabbitmq_ct_helpers",
     beam_files = [":beam_files"],
+    extra_apps = [
+        "common_test",
+        "eunit",
+        "inet_tcp_proxy",
+        "inets",
+    ],
     license_files = [":license_files"],
     priv = [":priv"],
     deps = [

deps/rabbitmq_jms_topic_exchange/BUILD.bazel

@@ -44,10 +44,10 @@ rabbitmq_app(
     license_files = [":license_files"],
     priv = [":priv"],
     deps = [
+        "//deps/khepri:erlang_app",
+        "//deps/khepri_mnesia_migration:erlang_app",
         "//deps/rabbit:erlang_app",
         "//deps/rabbit_common:erlang_app",
-        "@khepri//:erlang_app",
-        "@khepri_mnesia_migration//:erlang_app",
     ],
 )
 

deps/rabbitmq_jms_topic_exchange/app.bzl

@@ -19,10 +19,10 @@ def all_beam_files(name = "all_beam_files"):
         dest = "ebin",
         erlc_opts = "//:erlc_opts",
         deps = [
+            "//deps/khepri:erlang_app",
+            "//deps/khepri_mnesia_migration:erlang_app",
             "//deps/rabbit:erlang_app",
             "//deps/rabbit_common:erlang_app",
-            "@khepri//:erlang_app",
-            "@khepri_mnesia_migration//:erlang_app",
         ],
     )
 
@@ -46,10 +46,10 @@ def all_test_beam_files(name = "all_test_beam_files"):
         dest = "test",
         erlc_opts = "//:test_erlc_opts",
         deps = [
+            "//deps/khepri:erlang_app",
+            "//deps/khepri_mnesia_migration:erlang_app",
             "//deps/rabbit:erlang_app",
             "//deps/rabbit_common:erlang_app",
-            "@khepri//:erlang_app",
-            "@khepri_mnesia_migration//:erlang_app",
         ],
     )
 

deps/rabbitmq_mqtt/Makefile

@@ -94,7 +94,7 @@ define ct_master.erl
 	halt(0)
 endef
 
-PARALLEL_CT_SET_1_A = auth retainer
+PARALLEL_CT_SET_1_A = auth rabbit_mqtt_retained_msg_store retainer
 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/app.bzl

@@ -329,3 +329,12 @@ def test_suite_beam_files(name = "test_suite_beam_files"):
         erlc_opts = "//:test_erlc_opts",
         deps = ["//deps/amqp_client:erlang_app", "//deps/rabbitmq_ct_helpers:erlang_app"],
     )
+    erlang_bytecode(
+        name = "rabbit_mqtt_retained_msg_store_SUITE_beam_files",
+        testonly = True,
+        srcs = ["test/rabbit_mqtt_retained_msg_store_SUITE.erl"],
+        outs = ["test/rabbit_mqtt_retained_msg_store_SUITE.beam"],
+        hdrs = ["include/rabbit_mqtt_packet.hrl"],
+        app_name = "rabbitmq_mqtt",
+        erlc_opts = "//:test_erlc_opts",
+    )

deps/rabbitmq_mqtt/src/rabbit_mqtt_processor.erl

@@ -956,44 +956,52 @@ 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, TopicFilter, SubscribeQos, S)
+              end, State0, Msgs);
+        #mqtt_msg{} = SingleMsg ->
+            send_retained_message_to_client(SingleMsg, TopicFilter, SubscribeQos, State0)
     end.
 
+send_retained_message_to_client(#mqtt_msg{qos = MsgQos,
+                                        retain = Retain,
+                                        payload = Payload,
+                                        props = Props0},
+                               TopicFilter,
+                               SubscribeQos,
+                               State0 = #state{packet_id = PacketId0}) ->
+    Qos = effective_qos(MsgQos, SubscribeQos),
+    {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.
+
 clear_will_msg(#state{cfg = #cfg{vhost = Vhost,
                                  client_id = ClientId}} = State) ->
     QNameBin = rabbit_mqtt_util:queue_name_bin(ClientId, will),