add core tests for handling multiple channel interceptors

Author: Ayanda-D

deps/rabbit/test/channel_interceptor_SUITE.erl

@@ -24,7 +24,10 @@ groups() ->
           register_interceptor,
           register_interceptor_failing_with_amqp_error,
           register_interceptor_crashing_with_amqp_error_exception,
-          register_failing_interceptors
+          register_failing_interceptors,
+          multiple_interceptors_ordered_by_priority,
+          reject_interceptors_with_same_priority_for_same_operation,
+          priority_overridden_by_config
         ]}
     ].
 
@@ -213,6 +216,98 @@ register_failing_interceptors(Config) ->
     passed = rabbit_ct_broker_helpers:rpc(Config, 0,
       ?MODULE, register_interceptor1, [Config, failing_dummy_interceptor]).
 
+multiple_interceptors_ordered_by_priority(Config) ->
+    passed = rabbit_ct_broker_helpers:rpc(Config, 0,
+      ?MODULE, multiple_interceptors_ordered_by_priority1, [Config]).
+
+multiple_interceptors_ordered_by_priority1(Config) ->
+    Ch = rabbit_ct_client_helpers:open_channel(Config, 0),
+    QName = <<"multiple-interceptors-q">>,
+    #'queue.declare_ok'{} = amqp_channel:call(Ch, #'queue.declare'{queue = QName,
+                                                                    durable = true}),
+
+    ok = application:set_env(rabbit, channel_interceptor_priorities,
+                             [{dummy_interceptor_priority_1, 1},
+                              {dummy_interceptor_priority_2, 2},
+                              {dummy_interceptor_priority_3, 3}]),
+
+    ok = rabbit_registry:register(channel_interceptor,
+                                  <<"dummy interceptor priority 3">>, dummy_interceptor_priority_3),
+    ok = rabbit_registry:register(channel_interceptor,
+                                  <<"dummy interceptor priority 2">>, dummy_interceptor_priority_2),
+    ok = rabbit_registry:register(channel_interceptor,
+                                  <<"dummy interceptor priority 1">>, dummy_interceptor_priority_1),
+
+    %% Interceptors run in ascending priority order regardless of registration order,
+    %% so the payload becomes <<"foo1">>, then <<"foo12">>, then <<"foo123">>.
+    check_send_receive(Ch, QName, <<"foo">>, <<"foo123">>),
+
+    ok = rabbit_registry:unregister(channel_interceptor, <<"dummy interceptor priority 1">>),
+    ok = rabbit_registry:unregister(channel_interceptor, <<"dummy interceptor priority 2">>),
+    ok = rabbit_registry:unregister(channel_interceptor, <<"dummy interceptor priority 3">>),
+    ok = application:unset_env(rabbit, channel_interceptor_priorities),
+
+    #'queue.delete_ok'{} = amqp_channel:call(Ch, #'queue.delete'{queue = QName}),
+    passed.
+
+reject_interceptors_with_same_priority_for_same_operation(Config) ->
+    passed = rabbit_ct_broker_helpers:rpc(Config, 0,
+      ?MODULE, reject_interceptors_with_same_priority_for_same_operation1, [Config]).
+
+reject_interceptors_with_same_priority_for_same_operation1(_Config) ->
+    ok = rabbit_registry:register(channel_interceptor,
+                                  <<"dummy interceptor priority 1">>,
+                                  dummy_interceptor_priority_1),
+    ok = rabbit_registry:register(channel_interceptor,
+                                  <<"dummy interceptor priority 1 conflict">>,
+                                  dummy_interceptor_priority_1_conflict),
+    try
+        %% Initialising interceptors must fail: two interceptors with the same
+        %% priority handle the same AMQP operation.
+        rabbit_channel_interceptor:init(self())
+    catch
+        exit:{amqp_error, internal_error, _, _} -> ok
+    after
+        rabbit_registry:unregister(channel_interceptor, <<"dummy interceptor priority 1">>),
+        rabbit_registry:unregister(channel_interceptor, <<"dummy interceptor priority 1 conflict">>)
+    end,
+    passed.
+
+priority_overridden_by_config(Config) ->
+    passed = rabbit_ct_broker_helpers:rpc(Config, 0,
+      ?MODULE, priority_overridden_by_config1, [Config]).
+
+priority_overridden_by_config1(Config) ->
+    Ch = rabbit_ct_client_helpers:open_channel(Config, 0),
+    QName = <<"priority-override-q">>,
+    #'queue.declare_ok'{} = amqp_channel:call(Ch, #'queue.declare'{queue = QName,
+                                                                    durable = true}),
+
+    %% priority_1 (config priority=1) runs before priority_3 (config priority=3),
+    %% so the result is <<"foo13">>.
+    ok = application:set_env(rabbit, channel_interceptor_priorities,
+                             [{dummy_interceptor_priority_1, 1},
+                              {dummy_interceptor_priority_3, 3}]),
+    ok = rabbit_registry:register(channel_interceptor,
+                                  <<"dummy interceptor priority 1">>, dummy_interceptor_priority_1),
+    ok = rabbit_registry:register(channel_interceptor,
+                                  <<"dummy interceptor priority 3">>, dummy_interceptor_priority_3),
+    check_send_receive(Ch, QName, <<"foo">>, <<"foo13">>),
+
+    %% Reconfigure priority_3 to run first (priority=0). Now the result is <<"foo31">>.
+    ok = application:set_env(rabbit, channel_interceptor_priorities,
+                             [{dummy_interceptor_priority_1, 1},
+                              {dummy_interceptor_priority_3, 0}]),
+    rabbit_channel:refresh_interceptors(),
+    check_send_receive(Ch, QName, <<"foo">>, <<"foo31">>),
+
+    ok = application:unset_env(rabbit, channel_interceptor_priorities),
+    ok = rabbit_registry:unregister(channel_interceptor, <<"dummy interceptor priority 1">>),
+    ok = rabbit_registry:unregister(channel_interceptor, <<"dummy interceptor priority 3">>),
+
+    #'queue.delete_ok'{} = amqp_channel:call(Ch, #'queue.delete'{queue = QName}),
+    passed.
+
 check_send_receive(Ch1, QName, Send, Receive) ->
     amqp_channel:call(Ch1,
                         #'basic.publish'{routing_key = QName},

deps/rabbit/test/dummy_interceptor_priority_1.erl

@@ -0,0 +1,27 @@
+%% This Source Code Form is subject to the terms of the Mozilla Public
+%% License, v. 2.0. If a copy of the MPL was not distributed with this
+%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
+%%
+%% Copyright (c) 2007-2026 Broadcom. All Rights Reserved. The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries. All rights reserved.
+%%
+
+-module(dummy_interceptor_priority_1).
+
+-behaviour(rabbit_channel_interceptor).
+
+-include_lib("rabbit_common/include/rabbit.hrl").
+-include_lib("rabbit_common/include/rabbit_framing.hrl").
+
+-compile(export_all).
+
+init(_Ch) -> undefined.
+
+description() -> [{description, <<"Appends '1' to basic.publish payload">>}].
+
+intercept(#'basic.publish'{} = Method, Content, _State) ->
+    Payload = iolist_to_binary(lists:reverse(Content#content.payload_fragments_rev)),
+    {Method, Content#content{payload_fragments_rev = [<<Payload/binary, "1">>]}};
+intercept(Method, Content, _State) ->
+    {Method, Content}.
+
+applies_to() -> ['basic.publish'].

deps/rabbit/test/dummy_interceptor_priority_1_conflict.erl

@@ -0,0 +1,24 @@
+%% This Source Code Form is subject to the terms of the Mozilla Public
+%% License, v. 2.0. If a copy of the MPL was not distributed with this
+%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
+%%
+%% Copyright (c) 2007-2026 Broadcom. All Rights Reserved. The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries. All rights reserved.
+%%
+
+-module(dummy_interceptor_priority_1_conflict).
+
+-behaviour(rabbit_channel_interceptor).
+
+-include_lib("rabbit_common/include/rabbit.hrl").
+-include_lib("rabbit_common/include/rabbit_framing.hrl").
+
+-compile(export_all).
+
+init(_Ch) -> undefined.
+
+description() -> [{description, <<"Conflicts with dummy_interceptor_priority_1">>}].
+
+intercept(Method, Content, _State) ->
+    {Method, Content}.
+
+applies_to() -> ['basic.publish'].

deps/rabbit/test/dummy_interceptor_priority_2.erl

@@ -0,0 +1,27 @@
+%% This Source Code Form is subject to the terms of the Mozilla Public
+%% License, v. 2.0. If a copy of the MPL was not distributed with this
+%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
+%%
+%% Copyright (c) 2007-2026 Broadcom. All Rights Reserved. The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries. All rights reserved.
+%%
+
+-module(dummy_interceptor_priority_2).
+
+-behaviour(rabbit_channel_interceptor).
+
+-include_lib("rabbit_common/include/rabbit.hrl").
+-include_lib("rabbit_common/include/rabbit_framing.hrl").
+
+-compile(export_all).
+
+init(_Ch) -> undefined.
+
+description() -> [{description, <<"Appends '2' to basic.publish payload">>}].
+
+intercept(#'basic.publish'{} = Method, Content, _State) ->
+    Payload = iolist_to_binary(lists:reverse(Content#content.payload_fragments_rev)),
+    {Method, Content#content{payload_fragments_rev = [<<Payload/binary, "2">>]}};
+intercept(Method, Content, _State) ->
+    {Method, Content}.
+
+applies_to() -> ['basic.publish'].

deps/rabbit/test/dummy_interceptor_priority_3.erl

@@ -0,0 +1,27 @@
+%% This Source Code Form is subject to the terms of the Mozilla Public
+%% License, v. 2.0. If a copy of the MPL was not distributed with this
+%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
+%%
+%% Copyright (c) 2007-2026 Broadcom. All Rights Reserved. The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries. All rights reserved.
+%%
+
+-module(dummy_interceptor_priority_3).
+
+-behaviour(rabbit_channel_interceptor).
+
+-include_lib("rabbit_common/include/rabbit.hrl").
+-include_lib("rabbit_common/include/rabbit_framing.hrl").
+
+-compile(export_all).
+
+init(_Ch) -> undefined.
+
+description() -> [{description, <<"Appends '3' to basic.publish payload">>}].
+
+intercept(#'basic.publish'{} = Method, Content, _State) ->
+    Payload = iolist_to_binary(lists:reverse(Content#content.payload_fragments_rev)),
+    {Method, Content#content{payload_fragments_rev = [<<Payload/binary, "3">>]}};
+intercept(Method, Content, _State) ->
+    {Method, Content}.
+
+applies_to() -> ['basic.publish'].