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
105
106
107
108
109
110
111
112
113
|
%% The contents of this file are subject to the Mozilla Public License
%% Version 1.1 (the "License"); you may not use this file except in
%% compliance with the License. You may obtain a copy of the License
%% at http://www.mozilla.org/MPL/
%%
%% Software distributed under the License is distributed on an "AS IS"
%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
%% the License for the specific language governing rights and
%% limitations under the License.
%%
%% The Original Code is RabbitMQ.
%%
%% The Initial Developer of the Original Code is GoPivotal, Inc.
%% Copyright (c) 2007-2013 GoPivotal, Inc. All rights reserved.
%%
-module(rabbit_direct).
-export([boot/0, force_event_refresh/0, list/0, connect/5,
start_channel/9, disconnect/2]).
%% Internal
-export([list_local/0]).
-include("rabbit.hrl").
%%----------------------------------------------------------------------------
-ifdef(use_specs).
-spec(boot/0 :: () -> 'ok').
-spec(force_event_refresh/0 :: () -> 'ok').
-spec(list/0 :: () -> [pid()]).
-spec(list_local/0 :: () -> [pid()]).
-spec(connect/5 :: ((rabbit_types:username() | rabbit_types:user() |
{rabbit_types:username(), rabbit_types:password()}),
rabbit_types:vhost(), rabbit_types:protocol(), pid(),
rabbit_event:event_props()) ->
rabbit_types:ok_or_error2(
{rabbit_types:user(), rabbit_framing:amqp_table()},
'broker_not_found_on_node' |
{'auth_failure', string()} | 'access_refused')).
-spec(start_channel/9 ::
(rabbit_channel:channel_number(), pid(), pid(), string(),
rabbit_types:protocol(), rabbit_types:user(), rabbit_types:vhost(),
rabbit_framing:amqp_table(), pid()) -> {'ok', pid()}).
-spec(disconnect/2 :: (pid(), rabbit_event:event_props()) -> 'ok').
-endif.
%%----------------------------------------------------------------------------
boot() -> rabbit_sup:start_supervisor_child(
rabbit_direct_client_sup, rabbit_client_sup,
[{local, rabbit_direct_client_sup},
{rabbit_channel_sup, start_link, []}]).
force_event_refresh() ->
[Pid ! force_event_refresh || Pid<- list()],
ok.
list_local() ->
pg_local:get_members(rabbit_direct).
list() ->
rabbit_misc:append_rpc_all_nodes(rabbit_mnesia:cluster_nodes(running),
rabbit_direct, list_local, []).
%%----------------------------------------------------------------------------
connect(User = #user{}, VHost, Protocol, Pid, Infos) ->
try rabbit_access_control:check_vhost_access(User, VHost) of
ok -> ok = pg_local:join(rabbit_direct, Pid),
rabbit_event:notify(connection_created, Infos),
{ok, {User, rabbit_reader:server_properties(Protocol)}}
catch
exit:#amqp_error{name = access_refused} ->
{error, access_refused}
end;
connect({Username, Password}, VHost, Protocol, Pid, Infos) ->
connect0(fun () -> rabbit_access_control:check_user_pass_login(
Username, Password) end,
VHost, Protocol, Pid, Infos);
connect(Username, VHost, Protocol, Pid, Infos) ->
connect0(fun () -> rabbit_access_control:check_user_login(
Username, []) end,
VHost, Protocol, Pid, Infos).
connect0(AuthFun, VHost, Protocol, Pid, Infos) ->
case rabbit:is_running() of
true -> case AuthFun() of
{ok, User} ->
connect(User, VHost, Protocol, Pid, Infos);
{refused, _M, _A} ->
{error, {auth_failure, "Refused"}}
end;
false -> {error, broker_not_found_on_node}
end.
start_channel(Number, ClientChannelPid, ConnPid, ConnName, Protocol, User,
VHost, Capabilities, Collector) ->
{ok, _, {ChannelPid, _}} =
supervisor2:start_child(
rabbit_direct_client_sup,
[{direct, Number, ClientChannelPid, ConnPid, ConnName, Protocol,
User, VHost, Capabilities, Collector}]),
{ok, ChannelPid}.
disconnect(Pid, Infos) ->
pg_local:leave(rabbit_direct, Pid),
rabbit_event:notify(connection_closed, Infos).
|
