Packages
amqp_client
3.0.2
4.2.1
4.2.0
retired
4.1.6
4.1.5
retired
4.1.5-rc.2
retired
4.1.5-rc.1
retired
4.1.5-1
4.0.3
4.0.3-rc.1
3.13.0-rc.2
3.13.0-rc.1
3.12.14
3.12.13
3.12.12
3.12.11
3.12.10
3.12.9
3.12.8
3.12.7
3.12.6
3.12.5
3.12.4
3.12.3
3.12.2
3.12.1
3.12.0
3.12.0-rc.4
3.11.28
3.11.27
3.11.26
3.11.25
3.11.24
3.11.23
3.11.22
3.11.21
3.11.20
3.11.19
3.11.18
3.11.17
3.11.13
3.11.12
3.11.11
3.11.10
3.11.9
3.11.8
3.11.7
3.11.6
3.11.5
3.11.4
3.11.3
3.11.2
3.11.1
3.11.0
3.11.0-rc.2
3.11.0-rc.1
3.11.0-1
3.10.20
3.10.19
3.10.18
3.10.17
3.10.16
3.10.15
3.10.14
3.10.13
3.10.12
3.10.11
3.10.10
3.10.9
3.10.8
3.10.7
3.10.6
3.10.5
3.10.4
3.10.3
3.10.2
3.10.1
3.10.0
3.10.0-rc.6
3.10.0-rc.5
3.9.29
3.9.28
3.9.27
3.9.26
3.9.25
3.9.24
3.9.23
3.9.22
3.9.21
3.9.20
3.9.19
3.9.18
3.9.17
3.9.16
3.9.15
3.9.11
3.9.8
3.9.5
3.9.4
3.9.1
3.8.35
3.8.34
3.8.33
3.8.32
3.8.31
3.8.30
3.8.25
3.8.21
3.8.19
3.8.14
3.8.11
3.8.10
3.8.10-rc.1
3.8.9
3.8.7
3.8.6
3.8.6-rc.2
3.8.6-rc.1
3.8.5
3.8.5-rc.2
3.8.5-rc.1
3.8.4
3.8.4-rc.3
3.8.4-rc.1
3.8.3
3.8.3-rc.2
3.8.3-rc.1
3.8.2
3.8.2-rc.1
3.8.1
3.8.1-rc.1
3.8.0
3.8.0-rc.3
3.8.0-rc.2
3.8.0-rc.1
3.7.28
3.7.27
3.7.27-rc.2
3.7.27-rc.1
3.7.26
3.7.25
3.7.25-rc.1
3.7.24
3.7.24-rc.2
3.7.24-rc.1
3.7.23
3.7.23-rc.1
3.7.22
3.7.22-rc.2
3.7.22-rc.1
3.7.21
3.7.20
3.7.20-rc.2
3.7.20-rc.1
3.7.19
3.7.18
3.7.18-rc.1
3.7.17
3.7.17-rc.3
3.7.17-rc.2
3.7.15
3.7.14
3.7.14-rc.2
3.7.14-rc.1
3.7.13
3.7.13-rc.2
3.7.13-rc.1
3.7.12
3.7.12-rc.2
3.7.12-rc.1
3.7.11
3.7.11-rc.2
3.7.11-rc.1
3.7.10-rc.4
3.7.10-rc.3
3.7.10-rc.2
3.7.10-rc.1
3.7.9
3.7.9-rc.3
3.7.9-rc.2
3.7.8
3.7.8-rc.4
3.7.8-rc.3
3.7.8-rc.2
3.7.8-rc.1
3.7.7
3.7.7-rc.2
3.7.7-rc.1
3.7.6
3.7.6-rc.2
3.7.6-rc.1
3.7.5
3.7.5-rc.1
3.7.4
3.7.4-rc.4
3.7.4-rc.3
3.7.4-rc.2
3.7.4-rc.1
3.7.3
3.7.3-rc.2
3.7.3-rc.1
3.7.2
3.7.1
3.7.0-rc.2
3.6.16
3.6.16-rc.1
3.6.15
3.6.15-rc.1
3.6.14
3.6.13
3.6.12
3.6.11
3.6.10
3.6.9
3.6.8
3.6.7
3.6.7-pre.1
3.5.6
3.5.0
3.4.0
3.3.5
3.0.2
0.0.0-rc.1
RabbitMQ AMQP Client
Current section
Files
Jump to
Current section
Files
src/amqp_direct_connection.erl
%% 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 VMware, Inc.
%% Copyright (c) 2007-2012 VMware, Inc. All rights reserved.
%%
%% @private
-module(amqp_direct_connection).
-include("amqp_client_internal.hrl").
-behaviour(amqp_gen_connection).
-export([server_close/3]).
-export([init/1, terminate/2, connect/4, do/2, open_channel_args/1, i/2,
info_keys/0, handle_message/2, closing/3, channels_terminated/1]).
-record(state, {node,
user,
vhost,
params,
adapter_info,
collector,
closing_reason %% undefined | Reason
}).
-define(INFO_KEYS, [type]).
-define(CREATION_EVENT_KEYS, [pid, protocol, host, port, name,
peer_host, peer_port,
user, vhost, client_properties, type]).
%%---------------------------------------------------------------------------
%% amqp_connection:close() logically closes from the client end. We may
%% want to close from the server end.
server_close(ConnectionPid, Code, Text) ->
Close = #'connection.close'{reply_text = Text,
reply_code = Code,
class_id = 0,
method_id = 0},
amqp_gen_connection:server_close(ConnectionPid, Close).
init([]) ->
{ok, #state{}}.
open_channel_args(#state{node = Node,
user = User,
vhost = VHost,
adapter_info = Info,
collector = Collector}) ->
[self(), Info#amqp_adapter_info.name, Node, User, VHost, Collector].
do(_Method, _State) ->
ok.
handle_message(force_event_refresh, State = #state{node = Node}) ->
rpc:call(Node, rabbit_event, notify,
[connection_created, connection_info(State)]),
{ok, State};
handle_message(closing_timeout, State = #state{closing_reason = Reason}) ->
{stop, {closing_timeout, Reason}, State};
handle_message(Msg, State) ->
{stop, {unexpected_msg, Msg}, State}.
closing(_ChannelCloseType, Reason, State) ->
{ok, State#state{closing_reason = Reason}}.
channels_terminated(State = #state{closing_reason = Reason,
collector = Collector}) ->
rabbit_queue_collector:delete_all(Collector),
{stop, {shutdown, Reason}, State}.
terminate(_Reason, #state{node = Node}) ->
rpc:call(Node, rabbit_direct, disconnect, [self(), [{pid, self()}]]),
ok.
i(type, _State) -> direct;
i(pid, _State) -> self();
%% AMQP Params
i(user, #state{params=#amqp_params_direct{username=#user{username=U}}}) -> U;
i(user, #state{params=#amqp_params_direct{username=U}}) -> U;
i(vhost, #state{params = P}) -> P#amqp_params_direct.virtual_host;
i(client_properties, #state{params = P}) ->
P#amqp_params_direct.client_properties;
%% Optional adapter info
i(protocol, #state{adapter_info = I}) -> I#amqp_adapter_info.protocol;
i(host, #state{adapter_info = I}) -> I#amqp_adapter_info.host;
i(port, #state{adapter_info = I}) -> I#amqp_adapter_info.port;
i(peer_host, #state{adapter_info = I}) -> I#amqp_adapter_info.peer_host;
i(peer_port, #state{adapter_info = I}) -> I#amqp_adapter_info.peer_port;
i(name, #state{adapter_info = I}) -> I#amqp_adapter_info.name;
i(Item, _State) -> throw({bad_argument, Item}).
info_keys() ->
?INFO_KEYS.
infos(Items, State) ->
[{Item, i(Item, State)} || Item <- Items].
connection_info(State = #state{adapter_info = I}) ->
infos(?CREATION_EVENT_KEYS, State) ++ I#amqp_adapter_info.additional_info.
connect(Params = #amqp_params_direct{username = Username,
password = Password,
node = Node,
adapter_info = Info,
virtual_host = VHost},
SIF, _ChMgr, State) ->
State1 = State#state{node = Node,
vhost = VHost,
params = Params,
adapter_info = ensure_adapter_info(Info)},
AuthToken = case Password of
none -> Username;
_ -> {Username, Password}
end,
case rpc:call(Node, rabbit_direct, connect,
[AuthToken, VHost, ?PROTOCOL, self(),
connection_info(State1)]) of
{ok, {User, ServerProperties}} ->
{ok, Collector} = SIF(),
State2 = State1#state{user = User,
collector = Collector},
{ok, {ServerProperties, 0, State2}};
{error, _} = E ->
E;
{badrpc, nodedown} ->
{error, {nodedown, Node}}
end.
ensure_adapter_info(none) ->
ensure_adapter_info(#amqp_adapter_info{});
ensure_adapter_info(A = #amqp_adapter_info{protocol = unknown}) ->
ensure_adapter_info(A#amqp_adapter_info{
protocol = {'Direct', ?PROTOCOL:version()}});
ensure_adapter_info(A = #amqp_adapter_info{name = unknown}) ->
Name = list_to_binary(rabbit_misc:pid_to_string(self())),
ensure_adapter_info(A#amqp_adapter_info{name = Name});
ensure_adapter_info(Info) -> Info.