%% 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-2017 Pivotal Software, Inc. All rights reserved. %% %% @type close_reason(Type) = {shutdown, amqp_reason(Type)}. %% @type amqp_reason(Type) = {Type, Code, Text} %% Code = non_neg_integer() %% Text = binary(). %% @doc This module is responsible for maintaining a connection to an AMQP %% broker and manages channels within the connection. This module is used to %% open and close connections to the broker as well as creating new channels %% within a connection.
%% The connections and channels created by this module are supervised under %% amqp_client's supervision tree. Please note that connections and channels %% do not get restarted automatically by the supervision tree in the case of a %% failure. If you need robust connections and channels, we recommend you use %% Erlang monitors on the returned connection and channel PIDs.
%%
%% In case of a failure or an AMQP error, the connection process exits with a %% meaningful exit reason:
%%
%% %% %% %% %% %% %% %% %% %% %% %% %% %% %% %% %% %% %% %% %% %% %% %% %% %% %% %% %% %%
CauseExit reason
Any reason, where Code would have been 200 otherwise```normal'''
User application calls amqp_connection:close/3```close_reason(app_initiated_close)'''
Server closes connection (hard error)```close_reason(server_initiated_close)'''
Server misbehaved (did not follow protocol)```close_reason(server_misbehaved)'''
AMQP client internal error - usually caused by a channel exiting %% with an unusual reason. This is usually accompanied by a more %% detailed error log from the channel```close_reason(internal_error)'''
Other error(various error reasons, causing more detailed logging)
%%
%% See type definitions below. -module(amqp_connection). -include("amqp_client_internal.hrl"). -export([open_channel/1, open_channel/2, open_channel/3, register_blocked_handler/2]). -export([start/1, start/2, close/1, close/2, close/3, close/4]). -export([error_atom/1]). -export([info/2, info_keys/1, info_keys/0]). -export([connection_name/1]). -export([socket_adapter_info/2]). -define(DEFAULT_CONSUMER, {amqp_selective_consumer, []}). -define(PROTOCOL_SSL_PORT, (?PROTOCOL_PORT - 1)). %%--------------------------------------------------------------------------- %% Type Definitions %%--------------------------------------------------------------------------- %% @type amqp_adapter_info() = #amqp_adapter_info{}. %% @type amqp_params_direct() = #amqp_params_direct{}. %% As defined in amqp_client.hrl. It contains the following fields: %% %% %% @type amqp_params_network() = #amqp_params_network{}. %% As defined in amqp_client.hrl. It contains the following fields: %% %%--------------------------------------------------------------------------- %% Starting a connection %%--------------------------------------------------------------------------- %% @spec (Params) -> {ok, Connection} | {error, Error} %% where %% Params = amqp_params_network() | amqp_params_direct() %% Connection = pid() %% @doc same as {@link amqp_connection:start/2. start(Params, undefined)} start(AmqpParams) -> start(AmqpParams, undefined). %% @spec (Params, ConnectionName) -> {ok, Connection} | {error, Error} %% where %% Params = amqp_params_network() | amqp_params_direct() %% ConnectionName = undefined | binary() %% Connection = pid() %% @doc Starts a connection to an AMQP server. Use network params to %% connect to a remote AMQP server or direct params for a direct %% connection to a RabbitMQ server, assuming that the server is %% running in the same process space. If the port is set to 'undefined', %% the default ports will be selected depending on whether this is a %% normal or an SSL connection. %% If ConnectionName is binary - it will be added to client_properties as %% user specified connection name. start(AmqpParams, ConnName) when ConnName == undefined; is_binary(ConnName) -> ensure_started(), AmqpParams0 = case AmqpParams of #amqp_params_direct{password = Password} -> AmqpParams#amqp_params_direct{password = credentials_obfuscation:encrypt(Password)}; #amqp_params_network{password = Password} -> AmqpParams#amqp_params_network{password = credentials_obfuscation:encrypt(Password)} end, AmqpParams1 = case AmqpParams0 of #amqp_params_network{port = undefined, ssl_options = none} -> AmqpParams0#amqp_params_network{port = ?PROTOCOL_PORT}; #amqp_params_network{port = undefined, ssl_options = _} -> AmqpParams0#amqp_params_network{port = ?PROTOCOL_SSL_PORT}; _ -> AmqpParams0 end, AmqpParams2 = set_connection_name(ConnName, AmqpParams1), AmqpParams3 = amqp_ssl:maybe_enhance_ssl_options(AmqpParams2), {ok, _Sup, Connection} = amqp_sup:start_connection_sup(AmqpParams3), amqp_gen_connection:connect(Connection). set_connection_name(undefined, Params) -> Params; set_connection_name(ConnName, #amqp_params_network{client_properties = Props} = Params) -> Params#amqp_params_network{ client_properties = [ {<<"connection_name">>, longstr, ConnName} | Props ]}; set_connection_name(ConnName, #amqp_params_direct{client_properties = Props} = Params) -> Params#amqp_params_direct{ client_properties = [ {<<"connection_name">>, longstr, ConnName} | Props ]}. %% Usually the amqp_client application will already be running. We %% check whether that is the case by invoking an undocumented function %% which does not require a synchronous call to the application %% controller. That way we don't risk a dead-lock if, say, the %% application controller is in the process of shutting down the very %% application which is making this call. ensure_started() -> [ensure_started(App) || App <- [syntax_tools, compiler, xmerl, rabbit_common, amqp_client, credentials_obfuscation]]. ensure_started(App) -> case is_pid(application_controller:get_master(App)) andalso amqp_sup:is_ready() of true -> ok; false -> case application:ensure_all_started(App) of {ok, _} -> ok; {error, _} = E -> throw(E) end end. %%--------------------------------------------------------------------------- %% Commands %%--------------------------------------------------------------------------- %% @doc Invokes open_channel(ConnectionPid, none, %% {amqp_selective_consumer, []}). Opens a channel without having to %% specify a channel number. This uses the default consumer %% implementation. open_channel(ConnectionPid) -> open_channel(ConnectionPid, none, ?DEFAULT_CONSUMER). %% @doc Invokes open_channel(ConnectionPid, none, Consumer). %% Opens a channel without having to specify a channel number. open_channel(ConnectionPid, {_, _} = Consumer) -> open_channel(ConnectionPid, none, Consumer); %% @doc Invokes open_channel(ConnectionPid, ChannelNumber, %% {amqp_selective_consumer, []}). Opens a channel, using the default %% consumer implementation. open_channel(ConnectionPid, ChannelNumber) when is_number(ChannelNumber) orelse ChannelNumber =:= none -> open_channel(ConnectionPid, ChannelNumber, ?DEFAULT_CONSUMER). %% @spec (ConnectionPid, ChannelNumber, Consumer) -> Result %% where %% ConnectionPid = pid() %% ChannelNumber = pos_integer() | 'none' %% Consumer = {ConsumerModule, ConsumerArgs} %% ConsumerModule = atom() %% ConsumerArgs = [any()] %% Result = {ok, ChannelPid} | {error, Error} %% ChannelPid = pid() %% @doc Opens an AMQP channel.
%% Opens a channel, using a proposed channel number and a specific consumer %% implementation.
%% ConsumerModule must implement the amqp_gen_consumer behaviour. ConsumerArgs %% is passed as parameter to ConsumerModule:init/1.
%% This function assumes that an AMQP connection (networked or direct) %% has already been successfully established.
%% ChannelNumber must be less than or equal to the negotiated %% max_channel value, or less than or equal to ?MAX_CHANNEL_NUMBER %% (65535) if the negotiated max_channel value is 0.
%% In the direct connection, max_channel is always 0. open_channel(ConnectionPid, ChannelNumber, {_ConsumerModule, _ConsumerArgs} = Consumer) -> amqp_gen_connection:open_channel(ConnectionPid, ChannelNumber, Consumer). %% @spec (ConnectionPid) -> ok | Error %% where %% ConnectionPid = pid() %% @doc Closes the channel, invokes %% close(Channel, 200, <<"Goodbye">>). close(ConnectionPid) -> close(ConnectionPid, 200, <<"Goodbye">>). %% @spec (ConnectionPid, Timeout) -> ok | Error %% where %% ConnectionPid = pid() %% Timeout = integer() %% @doc Closes the channel, using the supplied Timeout value. close(ConnectionPid, Timeout) -> close(ConnectionPid, 200, <<"Goodbye">>, Timeout). %% @spec (ConnectionPid, Code, Text) -> ok | closing %% where %% ConnectionPid = pid() %% Code = integer() %% Text = binary() %% @doc Closes the AMQP connection, allowing the caller to set the reply %% code and text. close(ConnectionPid, Code, Text) -> close(ConnectionPid, Code, Text, amqp_util:call_timeout()). %% @spec (ConnectionPid, Code, Text, Timeout) -> ok | closing %% where %% ConnectionPid = pid() %% Code = integer() %% Text = binary() %% Timeout = integer() %% @doc Closes the AMQP connection, allowing the caller to set the reply %% code and text, as well as a timeout for the operation, after which the %% connection will be abruptly terminated. close(ConnectionPid, Code, Text, Timeout) -> Close = #'connection.close'{reply_text = Text, reply_code = Code, class_id = 0, method_id = 0}, amqp_gen_connection:close(ConnectionPid, Close, Timeout). register_blocked_handler(ConnectionPid, BlockHandler) -> amqp_gen_connection:register_blocked_handler(ConnectionPid, BlockHandler). %%--------------------------------------------------------------------------- %% Other functions %%--------------------------------------------------------------------------- %% @spec (Code) -> atom() %% where %% Code = integer() %% @doc Returns a descriptive atom corresponding to the given AMQP %% error code. error_atom(Code) -> ?PROTOCOL:amqp_exception(Code). %% @spec (ConnectionPid, Items) -> ResultList %% where %% ConnectionPid = pid() %% Items = [Item] %% ResultList = [{Item, Result}] %% Item = atom() %% Result = term() %% @doc Returns information about the connection, as specified by the Items %% list. Item may be any atom returned by info_keys/1: %% info(ConnectionPid, Items) -> amqp_gen_connection:info(ConnectionPid, Items). %% @spec (ConnectionPid) -> Items %% where %% ConnectionPid = pid() %% Items = [Item] %% Item = atom() %% @doc Returns a list of atoms that can be used in conjunction with info/2. %% Note that the list differs from a type of connection to another (network vs. %% direct). Use info_keys/0 to get a list of info keys that can be used for %% any connection. info_keys(ConnectionPid) -> amqp_gen_connection:info_keys(ConnectionPid). %% @spec () -> Items %% where %% Items = [Item] %% Item = atom() %% @doc Returns a list of atoms that can be used in conjunction with info/2. %% These are general info keys, which can be used in any type of connection. %% Other info keys may exist for a specific type. To get the full list of %% atoms that can be used for a certain connection, use info_keys/1. info_keys() -> amqp_gen_connection:info_keys(). %% @doc Takes a socket and a protocol, returns an #amqp_adapter_info{} %% based on the socket for the protocol given. socket_adapter_info(Sock, Protocol) -> amqp_direct_connection:socket_adapter_info(Sock, Protocol). %% @spec (ConnectionPid) -> ConnectionName %% where %% ConnectionPid = pid() %% ConnectionName = binary() %% @doc Returns user specified connection name from client properties connection_name(ConnectionPid) -> ClientProperties = case info(ConnectionPid, [amqp_params]) of [{_, #amqp_params_network{client_properties = Props}}] -> Props; [{_, #amqp_params_direct{client_properties = Props}}] -> Props end, case lists:keyfind(<<"connection_name">>, 1, ClientProperties) of {<<"connection_name">>, _, ConnName} -> ConnName; false -> undefined end.