Support consistent hash exchange routing for MQTT 5.0

When a consistent hash exchange is bound to the MQTT topic exchange,
MQTT 5.0 messages can be routed to queues consistently based on the
Correlation-Data in the PUBLISH packet.

Author: ansd

deps/rabbit/src/mc_amqp.erl

@@ -95,7 +95,7 @@ x_header(Key, Msg) ->
 property(correlation_id, #msg{properties = #'v1_0.properties'{correlation_id = Corr}}) ->
     Corr;
 property(message_id, #msg{properties = #'v1_0.properties'{message_id = MsgId}}) ->
-     MsgId;
+    MsgId;
 property(_Prop, #msg{}) ->
     undefined.
 

deps/rabbitmq_mqtt/src/mc_mqtt.erl

@@ -213,6 +213,8 @@ size_prop(_, _, Sum) ->
 x_header(_Key, #mqtt_msg{}) ->
     undefined.
 
+property(correlation_id, #mqtt_msg{props = #{'Correlation-Data' := Corr}}) ->
+    {binary, Corr};
 property(_Key, #mqtt_msg{}) ->
     undefined.
 

deps/rabbitmq_mqtt/test/v5_SUITE.erl

@@ -128,7 +128,8 @@ cluster_size_1_tests() ->
      topic_alias_retained_message,
      topic_alias_disallowed_retained_message,
      extended_auth,
-     headers_exchange
+     headers_exchange,
+     consistent_hash_exchange
     ].
 
 cluster_size_3_tests() ->
@@ -2050,12 +2051,9 @@ headers_exchange(Config) ->
                               Ch, #'exchange.bind'{destination = HeadersX,
                                                    source = <<"amq.topic">>,
                                                    routing_key = <<"my.topic">>}),
-    #'queue.declare_ok'{} = amqp_channel:call(
-                              Ch, #'queue.declare'{queue = Q1,
-                                                   durable = true}),
-    #'queue.declare_ok'{} = amqp_channel:call(
-                              Ch, #'queue.declare'{queue = Q2,
-                                                   durable = true}),
+    [#'queue.declare_ok'{} = amqp_channel:call(
+                               Ch, #'queue.declare'{queue = Q,
+                                                    durable = true}) || Q <- Qs],
     #'queue.bind_ok'{} = amqp_channel:call(
                            Ch, #'queue.bind'{queue = Q1,
                                              exchange = HeadersX,
@@ -2107,6 +2105,72 @@ headers_exchange(Config) ->
     [#'queue.delete_ok'{} = amqp_channel:call(Ch, #'queue.delete'{queue = Q}) || Q <- Qs],
     ok = rabbit_ct_client_helpers:close_channels_and_connection(Config, 0).
 
+%% Binding a consistent hash exchange to the MQTT topic exchange should support
+%% consistent routing based on Correlation-Data in the PUBLISH packet.
+consistent_hash_exchange(Config) ->
+    ok = rabbit_ct_broker_helpers:enable_plugin(Config, 0, rabbitmq_consistent_hash_exchange),
+    HashX = <<"my-consistent-hash-exchange">>,
+    Q1 = <<"q1">>,
+    Q2 = <<"q2">>,
+    Qs = [Q1, Q2],
+    Ch = rabbit_ct_client_helpers:open_channel(Config),
+
+    #'exchange.declare_ok'{} = amqp_channel:call(
+                                 Ch, #'exchange.declare'{
+                                        exchange = HashX,
+                                        type = <<"x-consistent-hash">>,
+                                        arguments = [{<<"hash-property">>, longstr, <<"correlation_id">>}],
+                                        durable = true,
+                                        auto_delete = true}),
+    #'exchange.bind_ok'{} = amqp_channel:call(
+                              Ch, #'exchange.bind'{destination = HashX,
+                                                   source = <<"amq.topic">>,
+                                                   routing_key = <<"a.*">>}),
+    [#'queue.declare_ok'{} = amqp_channel:call(
+                               Ch, #'queue.declare'{queue = Q,
+                                                    durable = true}) || Q <- Qs],
+    [#'queue.bind_ok'{} = amqp_channel:call(
+                            Ch, #'queue.bind'{queue = Q,
+                                              exchange = HashX,
+                                              %% weight
+                                              routing_key = <<"1">>}) || Q <- Qs],
+
+    Rands = [integer_to_binary(rand:uniform(1000)) || _ <- lists:seq(1, 30)],
+    UniqRands = lists:uniq(Rands),
+    NumMsgs = 150,
+    C = connect(?FUNCTION_NAME, Config),
+    [begin
+         N = integer_to_binary(rand:uniform(1_000_000)),
+         Topic = <<"a/", N/binary>>,
+         {ok, _} =  emqtt:publish(C, Topic,
+                                  #{'Correlation-Data' => lists:nth(rand:uniform(length(UniqRands)), UniqRands)},
+                                  N, [{qos, 1}])
+     end || _ <- lists:seq(1, NumMsgs)],
+
+    #'basic.consume_ok'{consumer_tag = Ctag1} = amqp_channel:subscribe(
+                                                  Ch, #'basic.consume'{queue = Q1,
+                                                                       no_ack = true}, self()),
+    #'basic.consume_ok'{consumer_tag = Ctag2} = amqp_channel:subscribe(
+                                                  Ch, #'basic.consume'{queue = Q2,
+                                                                       no_ack = true}, self()),
+    {N1, Corrs1} = receive_correlations(Ctag1, 0, sets:new([{version, 2}])),
+    {N2, Corrs2} = receive_correlations(Ctag2, 0, sets:new([{version, 2}])),
+    ct:pal("q1: ~b messages, ~b unique correlation-data", [N1, sets:size(Corrs1)]),
+    ct:pal("q2: ~b messages, ~b unique correlation-data", [N2, sets:size(Corrs2)]),
+    %% All messages should be routed.
+    ?assertEqual(NumMsgs, N1 + N2),
+    %% Each of the 2 queues should have received at least 1 message.
+    ?assert(sets:size(Corrs1) > 0),
+    ?assert(sets:size(Corrs2) > 0),
+    %% Assert that the consistent hash exchange routed the given Correlation-Data consistently.
+    %% The same Correlation-Data should never be present in both queues.
+    Intersection = sets:intersection(Corrs1, Corrs2),
+    ?assert(sets:is_empty(Intersection)),
+
+    ok = emqtt:disconnect(C),
+    [#'queue.delete_ok'{} = amqp_channel:call(Ch, #'queue.delete'{queue = Q}) || Q <- Qs],
+    ok = rabbit_ct_client_helpers:close_channels_and_connection(Config, 0).
+
 %% -------------------------------------------------------------------
 %% Helpers
 %% -------------------------------------------------------------------