1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
|
%% 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-2021 VMware, Inc. or its affiliates. All rights reserved.
%%
-module(rabbit_channel_interceptor).
-include("rabbit_framing.hrl").
-include("rabbit.hrl").
-export([init/1, intercept_in/3]).
-behaviour(rabbit_registry_class).
-export([added_to_rabbit_registry/2, removed_from_rabbit_registry/1]).
-type(method_name() :: rabbit_framing:amqp_method_name()).
-type(original_method() :: rabbit_framing:amqp_method_record()).
-type(processed_method() :: rabbit_framing:amqp_method_record()).
-type(original_content() :: rabbit_types:maybe(rabbit_types:content())).
-type(processed_content() :: rabbit_types:maybe(rabbit_types:content())).
-type(interceptor_state() :: term()).
-callback description() -> [proplists:property()].
%% Derive some initial state from the channel. This will be passed back
%% as the third argument of intercept/3.
-callback init(rabbit_channel:channel()) -> interceptor_state().
-callback intercept(original_method(), original_content(),
interceptor_state()) ->
{processed_method(), processed_content()} |
rabbit_misc:channel_or_connection_exit().
-callback applies_to() -> list(method_name()).
added_to_rabbit_registry(_Type, _ModuleName) ->
rabbit_channel:refresh_interceptors().
removed_from_rabbit_registry(_Type) ->
rabbit_channel:refresh_interceptors().
init(Ch) ->
Mods = [M || {_, M} <- rabbit_registry:lookup_all(channel_interceptor)],
check_no_overlap(Mods),
[{Mod, Mod:init(Ch)} || Mod <- Mods].
check_no_overlap(Mods) ->
check_no_overlap1([sets:from_list(Mod:applies_to()) || Mod <- Mods]).
%% Check no non-empty pairwise intersection in a list of sets
check_no_overlap1(Sets) ->
lists:foldl(fun(Set, Union) ->
Is = sets:intersection(Set, Union),
case sets:size(Is) of
0 -> ok;
_ ->
internal_error("Interceptor: more than one "
"module handles ~p~n", [Is])
end,
sets:union(Set, Union)
end,
sets:new(),
Sets),
ok.
intercept_in(M, C, Mods) ->
lists:foldl(fun({Mod, ModState}, {M1, C1}) ->
call_module(Mod, ModState, M1, C1)
end,
{M, C},
Mods).
call_module(Mod, St, M, C) ->
% this little dance is because Mod might be unloaded at any point
case (catch {ok, Mod:intercept(M, C, St)}) of
{ok, R} -> validate_response(Mod, M, C, R);
{'EXIT', {undef, [{Mod, intercept, _, _} | _]}} -> {M, C}
end.
validate_response(Mod, M1, C1, R = {M2, C2}) ->
case {validate_method(M1, M2), validate_content(C1, C2)} of
{true, true} -> R;
{false, _} ->
internal_error("Interceptor: ~p expected to return "
"method: ~p but returned: ~p",
[Mod, rabbit_misc:method_record_type(M1),
rabbit_misc:method_record_type(M2)]);
{_, false} ->
internal_error("Interceptor: ~p expected to return "
"content iff content is provided but "
"content in = ~p; content out = ~p",
[Mod, C1, C2])
end.
validate_method(M, M2) ->
rabbit_misc:method_record_type(M) =:= rabbit_misc:method_record_type(M2).
validate_content(none, none) -> true;
validate_content(#content{}, #content{}) -> true;
validate_content(_, _) -> false.
%% keep dialyzer happy
-spec internal_error(string(), [any()]) -> no_return().
internal_error(Format, Args) ->
rabbit_misc:protocol_error(internal_error, Format, Args).
|
