root
/
rabbitmq-server
mirror of https://github.com/rabbitmq/rabbitmq-server.git
1
0
Fork 0
Code Issues Actions 1 Packages Projects Releases Wiki Activity

Merge bug24924

This commit is contained in:
Simon MacMullen 2012-06-01 11:00:49 +01:00
parent f4e69068cd 5128463a76
commit e07f7d6a50
32 changed files with 1144 additions and 981 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
docs/rabbitmqctl.1.xml
View File

@ -720,7 +720,7 @@
</varlistentry> </varlistentry>
<varlistentry> <varlistentry>
<term><cmdsynopsis><command>list_user_permissions</command> <arg choice="opt">-p <replaceable>vhostpath</replaceable></arg> <arg choice="req"><replaceable>username</replaceable></arg></cmdsynopsis></term> <term><cmdsynopsis><command>list_user_permissions</command> <arg choice="req"><replaceable>username</replaceable></arg></cmdsynopsis></term>
<listitem> <listitem>
<variablelist> <variablelist>
<varlistentry> <varlistentry>
@ -978,7 +978,8 @@
</varlistentry> </varlistentry>
<varlistentry> <varlistentry>
<term>type</term> <term>type</term>
<listitem><para>The exchange type (one of [<command>direct</command>, <listitem><para>The exchange type (such as
[<command>direct</command>,
<command>topic</command>, <command>headers</command>, <command>topic</command>, <command>headers</command>,
<command>fanout</command>]).</para></listitem> <command>fanout</command>]).</para></listitem>
</varlistentry> </varlistentry>

2
packaging/debs/Debian/debian/control
View File

@ -2,7 +2,7 @@ Source: rabbitmq-server
Section: net Section: net
Priority: extra Priority: extra
Maintainer: RabbitMQ Team <packaging@rabbitmq.com> Maintainer: RabbitMQ Team <packaging@rabbitmq.com>
Uploader: Emile Joubert <emile@rabbitmq.com> Uploaders: Emile Joubert <emile@rabbitmq.com>
DM-Upload-Allowed: yes DM-Upload-Allowed: yes
Build-Depends: cdbs, debhelper (>= 5), erlang-dev, python-simplejson, xmlto, xsltproc, erlang-nox (>= 1:12.b.3), erlang-src (>= 1:12.b.3), unzip, zip Build-Depends: cdbs, debhelper (>= 5), erlang-dev, python-simplejson, xmlto, xsltproc, erlang-nox (>= 1:12.b.3), erlang-src (>= 1:12.b.3), unzip, zip
Standards-Version: 3.8.0 Standards-Version: 3.8.0

2
scripts/rabbitmq-plugins
View File

@ -31,7 +31,7 @@ exec erl \
-noinput \ -noinput \
-hidden \ -hidden \
-sname rabbitmq-plugins$$ \ -sname rabbitmq-plugins$$ \
-s rabbit_plugins \ -s rabbit_plugins_main \
-enabled_plugins_file "$RABBITMQ_ENABLED_PLUGINS_FILE" \ -enabled_plugins_file "$RABBITMQ_ENABLED_PLUGINS_FILE" \
-plugins_dist_dir "$RABBITMQ_PLUGINS_DIR" \ -plugins_dist_dir "$RABBITMQ_PLUGINS_DIR" \
-extra "$@" -extra "$@"

2
scripts/rabbitmq-plugins.bat
View File

@ -47,7 +47,7 @@ if "!RABBITMQ_PLUGINS_DIR!"=="" (
set RABBITMQ_PLUGINS_DIR=!TDP0!..\plugins set RABBITMQ_PLUGINS_DIR=!TDP0!..\plugins
) )
"!ERLANG_HOME!\bin\erl.exe" -pa "!TDP0!..\ebin" -noinput -hidden -sname rabbitmq-plugins!RANDOM! -s rabbit_plugins -enabled_plugins_file "!RABBITMQ_ENABLED_PLUGINS_FILE!" -plugins_dist_dir "!RABBITMQ_PLUGINS_DIR:\=/!" -extra !STAR! "!ERLANG_HOME!\bin\erl.exe" -pa "!TDP0!..\ebin" -noinput -hidden -sname rabbitmq-plugins!RANDOM! -s rabbit_plugins_main -enabled_plugins_file "!RABBITMQ_ENABLED_PLUGINS_FILE!" -plugins_dist_dir "!RABBITMQ_PLUGINS_DIR:\=/!" -extra !STAR!
endlocal endlocal
endlocal endlocal

34
scripts/rabbitmq-server
View File

@ -58,7 +58,8 @@ DEFAULT_NODE_PORT=5672
##--- End of overridden <var_name> variables ##--- End of overridden <var_name> variables
RABBITMQ_START_RABBIT= RABBITMQ_START_RABBIT=
[ "x" = "x$RABBITMQ_ALLOW_INPUT" ] && RABBITMQ_START_RABBIT='-noinput' [ "x" = "x$RABBITMQ_ALLOW_INPUT" ] && RABBITMQ_START_RABBIT=" -noinput"
[ "x" = "x$RABBITMQ_NODE_ONLY" ] && RABBITMQ_START_RABBIT="$RABBITMQ_START_RABBIT -s rabbit boot "
case "$(uname -s)" in case "$(uname -s)" in
CYGWIN*) # we make no attempt to record the cygwin pid; rabbitmqctl wait CYGWIN*) # we make no attempt to record the cygwin pid; rabbitmqctl wait
@ -70,24 +71,16 @@ case "$(uname -s)" in
esac esac
RABBITMQ_EBIN_ROOT="${RABBITMQ_HOME}/ebin" RABBITMQ_EBIN_ROOT="${RABBITMQ_HOME}/ebin"
if [ "x" = "x$RABBITMQ_NODE_ONLY" ]; then if ! erl -pa "$RABBITMQ_EBIN_ROOT" \
if erl \ -noinput \
-pa "$RABBITMQ_EBIN_ROOT" \ -hidden \
-noinput \ -s rabbit_prelaunch \
-hidden \ -sname rabbitmqprelaunch$$ \
-s rabbit_prelaunch \ -extra "${RABBITMQ_NODENAME}";
-sname rabbitmqprelaunch$$ \
-extra "$RABBITMQ_ENABLED_PLUGINS_FILE" "$RABBITMQ_PLUGINS_DIR" "${RABBITMQ_PLUGINS_EXPAND_DIR}" "${RABBITMQ_NODENAME}"
then then
RABBITMQ_BOOT_FILE="${RABBITMQ_PLUGINS_EXPAND_DIR}/rabbit" exit 1;
RABBITMQ_EBIN_PATH=""
else
exit 1
fi
else
RABBITMQ_BOOT_FILE=start_sasl
RABBITMQ_EBIN_PATH="-pa ${RABBITMQ_EBIN_ROOT}"
fi fi
RABBITMQ_CONFIG_ARG= RABBITMQ_CONFIG_ARG=
[ -f "${RABBITMQ_CONFIG_FILE}.config" ] && RABBITMQ_CONFIG_ARG="-config ${RABBITMQ_CONFIG_FILE}" [ -f "${RABBITMQ_CONFIG_FILE}.config" ] && RABBITMQ_CONFIG_ARG="-config ${RABBITMQ_CONFIG_FILE}"
@ -100,10 +93,10 @@ RABBITMQ_LISTEN_ARG=
set -f set -f
exec erl \ exec erl \
${RABBITMQ_EBIN_PATH} \ -pa ${RABBITMQ_EBIN_ROOT} \
${RABBITMQ_START_RABBIT} \ ${RABBITMQ_START_RABBIT} \
-sname ${RABBITMQ_NODENAME} \ -sname ${RABBITMQ_NODENAME} \
-boot ${RABBITMQ_BOOT_FILE} \ -boot start_sasl \
${RABBITMQ_CONFIG_ARG} \ ${RABBITMQ_CONFIG_ARG} \
+W w \ +W w \
${RABBITMQ_SERVER_ERL_ARGS} \ ${RABBITMQ_SERVER_ERL_ARGS} \
@ -112,6 +105,9 @@ exec erl \
-sasl sasl_error_logger false \ -sasl sasl_error_logger false \
-rabbit error_logger '{file,"'${RABBITMQ_LOGS}'"}' \ -rabbit error_logger '{file,"'${RABBITMQ_LOGS}'"}' \
-rabbit sasl_error_logger '{file,"'${RABBITMQ_SASL_LOGS}'"}' \ -rabbit sasl_error_logger '{file,"'${RABBITMQ_SASL_LOGS}'"}' \
-rabbit enabled_plugins_file "\"$RABBITMQ_ENABLED_PLUGINS_FILE\"" \
-rabbit plugins_dir "\"$RABBITMQ_PLUGINS_DIR\"" \
-rabbit plugins_expand_dir "\"$RABBITMQ_PLUGINS_EXPAND_DIR\"" \
-os_mon start_cpu_sup false \ -os_mon start_cpu_sup false \
-os_mon start_disksup false \ -os_mon start_disksup false \
-os_mon start_memsup false \ -os_mon start_memsup false \

24
scripts/rabbitmq-server.bat
View File

@ -93,21 +93,17 @@ if "!RABBITMQ_PLUGINS_DIR!"=="" (
set RABBITMQ_EBIN_ROOT=!TDP0!..\ebin set RABBITMQ_EBIN_ROOT=!TDP0!..\ebin
"!ERLANG_HOME!\bin\erl.exe" ^ "!ERLANG_HOME!\bin\erl.exe" ^
-pa "!RABBITMQ_EBIN_ROOT!" ^ -pa "!RABBITMQ_EBIN_ROOT!" ^
-noinput -hidden ^ -noinput -hidden ^
-s rabbit_prelaunch ^ -s rabbit_prelaunch ^
-sname rabbitmqprelaunch!RANDOM! ^ -sname rabbitmqprelaunch!RANDOM! ^
-extra "!RABBITMQ_ENABLED_PLUGINS_FILE:\=/!" ^ -extra "!RABBITMQ_NODENAME!"
"!RABBITMQ_PLUGINS_DIR:\=/!" ^
"!RABBITMQ_PLUGINS_EXPAND_DIR:\=/!" ^
"!RABBITMQ_NODENAME!"
set RABBITMQ_BOOT_FILE=!RABBITMQ_PLUGINS_EXPAND_DIR!\rabbit
if ERRORLEVEL 1 ( if ERRORLEVEL 1 (
exit /B 1 exit /B 1
) )
set RABBITMQ_EBIN_PATH= set RABBITMQ_EBIN_PATH="-pa !RABBITMQ_EBIN_ROOT!"
if "!RABBITMQ_CONFIG_FILE!"=="" ( if "!RABBITMQ_CONFIG_FILE!"=="" (
set RABBITMQ_CONFIG_FILE=!RABBITMQ_BASE!\rabbitmq set RABBITMQ_CONFIG_FILE=!RABBITMQ_BASE!\rabbitmq
@ -127,9 +123,10 @@ if not "!RABBITMQ_NODE_IP_ADDRESS!"=="" (
) )
"!ERLANG_HOME!\bin\erl.exe" ^ "!ERLANG_HOME!\bin\erl.exe" ^
!RABBITMQ_EBIN_PATH! ^ -pa "!RABBITMQ_EBIN_ROOT!" ^
-noinput ^ -noinput ^
-boot "!RABBITMQ_BOOT_FILE!" ^ -boot start_sasl ^
-s rabbit boot ^
!RABBITMQ_CONFIG_ARG! ^ !RABBITMQ_CONFIG_ARG! ^
-sname !RABBITMQ_NODENAME! ^ -sname !RABBITMQ_NODENAME! ^
+W w ^ +W w ^
@ -142,6 +139,9 @@ if not "!RABBITMQ_NODE_IP_ADDRESS!"=="" (
-sasl sasl_error_logger false ^ -sasl sasl_error_logger false ^
-rabbit error_logger {file,\""!LOGS:\=/!"\"} ^ -rabbit error_logger {file,\""!LOGS:\=/!"\"} ^
-rabbit sasl_error_logger {file,\""!SASL_LOGS:\=/!"\"} ^ -rabbit sasl_error_logger {file,\""!SASL_LOGS:\=/!"\"} ^
-rabbit enabled_plugins_file \""!RABBITMQ_ENABLED_PLUGINS_FILE:\=/!"\" ^
-rabbit plugins_dir \""!RABBITMQ_PLUGINS_DIR:\=/!"\" ^
-rabbit plugins_expand_dir \""!RABBITMQ_PLUGINS_EXPAND_DIR:\=/!"\" ^
-os_mon start_cpu_sup false ^ -os_mon start_cpu_sup false ^
-os_mon start_disksup false ^ -os_mon start_disksup false ^
-os_mon start_memsup false ^ -os_mon start_memsup false ^

24
scripts/rabbitmq-service.bat
View File

@ -157,22 +157,6 @@ if "!RABBITMQ_ENABLED_PLUGINS_FILE!"=="" (
set RABBITMQ_PLUGINS_DIR=!TDP0!..\plugins set RABBITMQ_PLUGINS_DIR=!TDP0!..\plugins
set RABBITMQ_EBIN_ROOT=!TDP0!..\ebin set RABBITMQ_EBIN_ROOT=!TDP0!..\ebin
"!ERLANG_HOME!\bin\erl.exe" ^
-pa "!RABBITMQ_EBIN_ROOT!" ^
-noinput -hidden ^
-s rabbit_prelaunch ^
-extra "!RABBITMQ_ENABLED_PLUGINS_FILE:\=/!" ^
"!RABBITMQ_PLUGINS_DIR:\=/!" ^
"!RABBITMQ_PLUGINS_EXPAND_DIR:\=/!" ^
""
set RABBITMQ_BOOT_FILE=!RABBITMQ_PLUGINS_EXPAND_DIR!\rabbit
if ERRORLEVEL 1 (
exit /B 1
)
set RABBITMQ_EBIN_PATH=
if "!RABBITMQ_CONFIG_FILE!"=="" ( if "!RABBITMQ_CONFIG_FILE!"=="" (
set RABBITMQ_CONFIG_FILE=!RABBITMQ_BASE!\rabbitmq set RABBITMQ_CONFIG_FILE=!RABBITMQ_BASE!\rabbitmq
) )
@ -191,8 +175,9 @@ if not "!RABBITMQ_NODE_IP_ADDRESS!"=="" (
) )
set ERLANG_SERVICE_ARGUMENTS= ^ set ERLANG_SERVICE_ARGUMENTS= ^
!RABBITMQ_EBIN_PATH! ^ -pa "!RABBITMQ_EBIN_ROOT!" ^
-boot "!RABBITMQ_BOOT_FILE!" ^ -boot start_sasl ^
-s rabbit boot ^
!RABBITMQ_CONFIG_ARG! ^ !RABBITMQ_CONFIG_ARG! ^
+W w ^ +W w ^
+A30 ^ +A30 ^
@ -204,6 +189,9 @@ set ERLANG_SERVICE_ARGUMENTS= ^
-sasl sasl_error_logger false ^ -sasl sasl_error_logger false ^
-rabbit error_logger {file,\""!LOGS:\=/!"\"} ^ -rabbit error_logger {file,\""!LOGS:\=/!"\"} ^
-rabbit sasl_error_logger {file,\""!SASL_LOGS:\=/!"\"} ^ -rabbit sasl_error_logger {file,\""!SASL_LOGS:\=/!"\"} ^
-rabbit enabled_plugins_file \""!RABBITMQ_ENABLED_PLUGINS_FILE:\=/!"\" ^
-rabbit plugins_dir \""!RABBITMQ_PLUGINS_DIR:\=/!"\" ^
-rabbit plugins_expand_dir \""!RABBITMQ_PLUGINS_EXPAND_DIR:\=/!"\" ^
-os_mon start_cpu_sup false ^ -os_mon start_cpu_sup false ^
-os_mon start_disksup false ^ -os_mon start_disksup false ^
-os_mon start_memsup false ^ -os_mon start_memsup false ^

2
scripts/rabbitmqctl
View File

@ -32,6 +32,6 @@ exec erl \
-hidden \ -hidden \
${RABBITMQ_CTL_ERL_ARGS} \ ${RABBITMQ_CTL_ERL_ARGS} \
-sname rabbitmqctl$$ \ -sname rabbitmqctl$$ \
-s rabbit_control \ -s rabbit_control_main \
-nodename $RABBITMQ_NODENAME \ -nodename $RABBITMQ_NODENAME \
-extra "$@" -extra "$@"

4
scripts/rabbitmqctl.bat
View File

@ -34,7 +34,7 @@ if "!RABBITMQ_NODENAME!"=="" (
if not exist "!ERLANG_HOME!\bin\erl.exe" ( if not exist "!ERLANG_HOME!\bin\erl.exe" (
echo. echo.
echo ****************************** echo ******************************
echo ERLANG_HOME not set correctly. echo ERLANG_HOME not set correctly.
echo ****************************** echo ******************************
echo. echo.
echo Please either set ERLANG_HOME to point to your Erlang installation or place the echo Please either set ERLANG_HOME to point to your Erlang installation or place the
@ -43,7 +43,7 @@ if not exist "!ERLANG_HOME!\bin\erl.exe" (
exit /B exit /B
) )
"!ERLANG_HOME!\bin\erl.exe" -pa "!TDP0!..\ebin" -noinput -hidden !RABBITMQ_CTL_ERL_ARGS! -sname rabbitmqctl!RANDOM! -s rabbit_control -nodename !RABBITMQ_NODENAME! -extra !STAR! "!ERLANG_HOME!\bin\erl.exe" -pa "!TDP0!..\ebin" -noinput -hidden !RABBITMQ_CTL_ERL_ARGS! -sname rabbitmqctl!RANDOM! -s rabbit_control_main -nodename !RABBITMQ_NODENAME! -extra !STAR!
endlocal endlocal
endlocal endlocal

121
src/app_utils.erl Normal file
View File

@ -0,0 +1,121 @@
%% 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.
%%
-module(app_utils).
-export([load_applications/1, start_applications/1,
stop_applications/1, app_dependency_order/2,
wait_for_applications/1]).
-ifdef(use_specs).
-spec load_applications([atom()]) -> 'ok'.
-spec start_applications([atom()]) -> 'ok'.
-spec stop_applications([atom()]) -> 'ok'.
-spec wait_for_applications([atom()]) -> 'ok'.
-spec app_dependency_order([atom()], boolean()) -> [digraph:vertex()].
-endif.
%%---------------------------------------------------------------------------
%% Public API
load_applications(Apps) ->
load_applications(queue:from_list(Apps), sets:new()),
ok.
start_applications(Apps) ->
manage_applications(fun lists:foldl/3,
fun application:start/1,
fun application:stop/1,
already_started,
cannot_start_application,
Apps).
stop_applications(Apps) ->
manage_applications(fun lists:foldr/3,
fun application:stop/1,
fun application:start/1,
not_started,
cannot_stop_application,
Apps).
wait_for_applications(Apps) ->
[wait_for_application(App) || App <- Apps], ok.
app_dependency_order(RootApps, StripUnreachable) ->
{ok, G} = rabbit_misc:build_acyclic_graph(
fun (App, _Deps) -> [{App, App}] end,
fun (App, Deps) -> [{Dep, App} || Dep <- Deps] end,
[{App, app_dependencies(App)} ||
{App, _Desc, _Vsn} <- application:loaded_applications()]),
try
case StripUnreachable of
true -> digraph:del_vertices(G, digraph:vertices(G) --
digraph_utils:reachable(RootApps, G));
false -> ok
end,
digraph_utils:topsort(G)
after
true = digraph:delete(G)
end.
%%---------------------------------------------------------------------------
%% Private API
wait_for_application(Application) ->
case lists:keymember(Application, 1, application:which_applications()) of
true -> ok;
false -> timer:sleep(1000),
wait_for_application(Application)
end.
load_applications(Worklist, Loaded) ->
case queue:out(Worklist) of
{empty, _WorkList} ->
ok;
{{value, App}, Worklist1} ->
case sets:is_element(App, Loaded) of
true -> load_applications(Worklist1, Loaded);
false -> case application:load(App) of
ok -> ok;
{error, {already_loaded, App}} -> ok;
Error -> throw(Error)
end,
load_applications(
queue:join(Worklist1,
queue:from_list(app_dependencies(App))),
sets:add_element(App, Loaded))
end
end.
app_dependencies(App) ->
case application:get_key(App, applications) of
undefined -> [];
{ok, Lst} -> Lst
end.
manage_applications(Iterate, Do, Undo, SkipError, ErrorTag, Apps) ->
Iterate(fun (App, Acc) ->
case Do(App) of
ok -> [App | Acc];
{error, {SkipError, _}} -> Acc;
{error, Reason} ->
lists:foreach(Undo, Acc),
throw({error, {ErrorTag, App, Reason}})
end
end, [], Apps),
ok.

189
src/gm.erl
View File

@ -433,51 +433,47 @@
-spec(confirmed_broadcast/2 :: (pid(), any()) -> 'ok'). -spec(confirmed_broadcast/2 :: (pid(), any()) -> 'ok').
-spec(group_members/1 :: (pid()) -> [pid()]). -spec(group_members/1 :: (pid()) -> [pid()]).
%% The joined, members_changed and handle_msg callbacks can all %% The joined, members_changed and handle_msg callbacks can all return
%% return any of the following terms: %% any of the following terms:
%% %%
%% 'ok' - the callback function returns normally %% 'ok' - the callback function returns normally
%% %%
%% {'stop', Reason} - the callback indicates the member should %% {'stop', Reason} - the callback indicates the member should stop
%% stop with reason Reason and should leave the group. %% with reason Reason and should leave the group.
%% %%
%% {'become', Module, Args} - the callback indicates that the %% {'become', Module, Args} - the callback indicates that the callback
%% callback module should be changed to Module and that the %% module should be changed to Module and that the callback functions
%% callback functions should now be passed the arguments %% should now be passed the arguments Args. This allows the callback
%% Args. This allows the callback module to be dynamically %% module to be dynamically changed.
%% changed.
%% Called when we've successfully joined the group. Supplied with %% Called when we've successfully joined the group. Supplied with Args
%% Args provided in start_link, plus current group members. %% provided in start_link, plus current group members.
-callback joined(Args :: term(), Members :: [pid()]) -> -callback joined(Args :: term(), Members :: [pid()]) ->
ok | {stop, Reason :: term()} | {become, Module :: atom(), Args :: any()}. ok | {stop, Reason :: term()} | {become, Module :: atom(), Args :: any()}.
%% Supplied with Args provided in start_link, the list of new %% Supplied with Args provided in start_link, the list of new members
%% members and the list of members previously known to us that %% and the list of members previously known to us that have since
%% have since died. Note that if a member joins and dies very %% died. Note that if a member joins and dies very quickly, it's
%% quickly, it's possible that we will never see that member %% possible that we will never see that member appear in either births
%% appear in either births or deaths. However we are guaranteed %% or deaths. However we are guaranteed that (1) we will see a member
%% that (1) we will see a member joining either in the births %% joining either in the births here, or in the members passed to
%% here, or in the members passed to joined/2 before receiving %% joined/2 before receiving any messages from it; and (2) we will not
%% any messages from it; and (2) we will not see members die that %% see members die that we have not seen born (or supplied in the
%% we have not seen born (or supplied in the members to %% members to joined/2).
%% joined/2).
-callback members_changed(Args :: term(), Births :: [pid()], -callback members_changed(Args :: term(), Births :: [pid()],
Deaths :: [pid()]) -> Deaths :: [pid()]) ->
ok | {stop, Reason :: term()} | {become, Module :: atom(), Args :: any()}. ok | {stop, Reason :: term()} | {become, Module :: atom(), Args :: any()}.
%% Supplied with Args provided in start_link, the sender, and the %% Supplied with Args provided in start_link, the sender, and the
%% message. This does get called for messages injected by this %% message. This does get called for messages injected by this member,
%% member, however, in such cases, there is no special %% however, in such cases, there is no special significance of this
%% significance of this invocation: it does not indicate that the %% invocation: it does not indicate that the message has made it to
%% message has made it to any other members, let alone all other %% any other members, let alone all other members.
%% members.
-callback handle_msg(Args :: term(), From :: pid(), Message :: term()) -> -callback handle_msg(Args :: term(), From :: pid(), Message :: term()) ->
ok | {stop, Reason :: term()} | {become, Module :: atom(), Args :: any()}. ok | {stop, Reason :: term()} | {become, Module :: atom(), Args :: any()}.
%% Called on gm member termination as per rules in gen_server, %% Called on gm member termination as per rules in gen_server, with
%% with the Args provided in start_link plus the termination %% the Args provided in start_link plus the termination Reason.
%% Reason.
-callback terminate(Args :: term(), Reason :: term()) -> -callback terminate(Args :: term(), Reason :: term()) ->
ok | term(). ok | term().
@ -533,7 +529,7 @@ init([GroupName, Module, Args]) ->
group_name = GroupName, group_name = GroupName,
module = Module, module = Module,
view = undefined, view = undefined,
pub_count = 0, pub_count = -1,
members_state = undefined, members_state = undefined,
callback_args = Args, callback_args = Args,
confirms = queue:new(), confirms = queue:new(),
@ -575,33 +571,39 @@ handle_call({add_on_right, NewMember}, _From,
members_state = MembersState, members_state = MembersState,
module = Module, module = Module,
callback_args = Args }) -> callback_args = Args }) ->
Group = record_new_member_in_group( {MembersState1, Group} =
GroupName, Self, NewMember, record_new_member_in_group(
fun (Group1) -> GroupName, Self, NewMember,
View1 = group_to_view(Group1), fun (Group1) ->
ok = send_right(NewMember, View1, View1 = group_to_view(Group1),
{catchup, Self, prepare_members_state( MembersState1 = remove_erased_members(MembersState, View1),
MembersState)}) ok = send_right(NewMember, View1,
end), {catchup, Self,
prepare_members_state(MembersState1)}),
MembersState1
end),
View2 = group_to_view(Group), View2 = group_to_view(Group),
State1 = check_neighbours(State #state { view = View2 }), State1 = check_neighbours(State #state { view = View2,
members_state = MembersState1 }),
Result = callback_view_changed(Args, Module, View, View2), Result = callback_view_changed(Args, Module, View, View2),
handle_callback_result({Result, {ok, Group}, State1}). handle_callback_result({Result, {ok, Group}, State1}).
handle_cast({?TAG, ReqVer, Msg}, handle_cast({?TAG, ReqVer, Msg},
State = #state { view = View, State = #state { view = View,
members_state = MembersState,
group_name = GroupName, group_name = GroupName,
module = Module, module = Module,
callback_args = Args }) -> callback_args = Args }) ->
{Result, State1} = {Result, State1} =
case needs_view_update(ReqVer, View) of case needs_view_update(ReqVer, View) of
true -> true -> View1 = group_to_view(read_group(GroupName)),
View1 = group_to_view(read_group(GroupName)), MemberState1 = remove_erased_members(MembersState, View1),
{callback_view_changed(Args, Module, View, View1), {callback_view_changed(Args, Module, View, View1),
check_neighbours(State #state { view = View1 })}; check_neighbours(
false -> State #state { view = View1,
{ok, State} members_state = MemberState1 })};
false -> {ok, State}
end, end,
handle_callback_result( handle_callback_result(
if_callback_success( if_callback_success(
@ -665,22 +667,21 @@ handle_info({'DOWN', MRef, process, _Pid, _Reason},
_ -> _ ->
View1 = View1 =
group_to_view(record_dead_member_in_group(Member, GroupName)), group_to_view(record_dead_member_in_group(Member, GroupName)),
State1 = State #state { view = View1 },
{Result, State2} = {Result, State2} =
case alive_view_members(View1) of case alive_view_members(View1) of
[Self] -> [Self] ->
maybe_erase_aliases( {Result1, State1} = maybe_erase_aliases(State, View1),
State1 #state { {Result1, State1 #state {
members_state = blank_member_state(), members_state = blank_member_state(),
confirms = purge_confirms(Confirms) }); confirms = purge_confirms(Confirms) }};
_ -> _ ->
%% here we won't be pointing out any deaths: %% here we won't be pointing out any deaths:
%% the concern is that there maybe births %% the concern is that there maybe births
%% which we'd otherwise miss. %% which we'd otherwise miss.
{callback_view_changed(Args, Module, View, View1), {callback_view_changed(Args, Module, View, View1),
State1} check_neighbours(State #state { view = View1 })}
end, end,
handle_callback_result({Result, check_neighbours(State2)}) handle_callback_result({Result, State2})
end. end.
@ -693,9 +694,13 @@ terminate(Reason, State = #state { module = Module,
code_change(_OldVsn, State, _Extra) -> code_change(_OldVsn, State, _Extra) ->
{ok, State}. {ok, State}.
prioritise_info(flush, _State) -> 1; prioritise_info(flush, _State) ->
prioritise_info({'DOWN', _MRef, process, _Pid, _Reason}, _State) -> 1; 1;
prioritise_info(_ , _State) -> 0. prioritise_info({'DOWN', _MRef, process, _Pid, _Reason},
#state { members_state = MS }) when MS /= undefined ->
1;
prioritise_info(_, _State) ->
0.
handle_msg(check_neighbours, State) -> handle_msg(check_neighbours, State) ->
@ -795,8 +800,8 @@ handle_msg({activity, Left, Activity},
State1 = State #state { members_state = MembersState1, State1 = State #state { members_state = MembersState1,
confirms = Confirms1 }, confirms = Confirms1 },
Activity3 = activity_finalise(Activity1), Activity3 = activity_finalise(Activity1),
{Result, State2} = maybe_erase_aliases(State1), ok = maybe_send_activity(Activity3, State1),
ok = maybe_send_activity(Activity3, State2), {Result, State2} = maybe_erase_aliases(State1, View),
if_callback_success( if_callback_success(
Result, fun activity_true/3, fun activity_false/3, Activity3, State2); Result, fun activity_true/3, fun activity_false/3, Activity3, State2);
@ -829,13 +834,14 @@ internal_broadcast(Msg, From, State = #state { self = Self,
confirms = Confirms, confirms = Confirms,
callback_args = Args, callback_args = Args,
broadcast_buffer = Buffer }) -> broadcast_buffer = Buffer }) ->
PubCount1 = PubCount + 1,
Result = Module:handle_msg(Args, get_pid(Self), Msg), Result = Module:handle_msg(Args, get_pid(Self), Msg),
Buffer1 = [{PubCount, Msg} | Buffer], Buffer1 = [{PubCount1, Msg} | Buffer],
Confirms1 = case From of Confirms1 = case From of
none -> Confirms; none -> Confirms;
_ -> queue:in({PubCount, From}, Confirms) _ -> queue:in({PubCount1, From}, Confirms)
end, end,
State1 = State #state { pub_count = PubCount + 1, State1 = State #state { pub_count = PubCount1,
confirms = Confirms1, confirms = Confirms1,
broadcast_buffer = Buffer1 }, broadcast_buffer = Buffer1 },
case From =/= none of case From =/= none of
@ -850,14 +856,17 @@ flush_broadcast_buffer(State = #state { broadcast_buffer = [] }) ->
State; State;
flush_broadcast_buffer(State = #state { self = Self, flush_broadcast_buffer(State = #state { self = Self,
members_state = MembersState, members_state = MembersState,
broadcast_buffer = Buffer }) -> broadcast_buffer = Buffer,
pub_count = PubCount }) ->
[{PubCount, _Msg}|_] = Buffer, %% ASSERTION match on PubCount
Pubs = lists:reverse(Buffer), Pubs = lists:reverse(Buffer),
Activity = activity_cons(Self, Pubs, [], activity_nil()), Activity = activity_cons(Self, Pubs, [], activity_nil()),
ok = maybe_send_activity(activity_finalise(Activity), State), ok = maybe_send_activity(activity_finalise(Activity), State),
MembersState1 = with_member( MembersState1 = with_member(
fun (Member = #member { pending_ack = PA }) -> fun (Member = #member { pending_ack = PA }) ->
PA1 = queue:join(PA, queue:from_list(Pubs)), PA1 = queue:join(PA, queue:from_list(Pubs)),
Member #member { pending_ack = PA1 } Member #member { pending_ack = PA1,
last_pub = PubCount }
end, Self, MembersState), end, Self, MembersState),
State #state { members_state = MembersState1, State #state { members_state = MembersState1,
broadcast_buffer = [] }. broadcast_buffer = [] }.
@ -1052,7 +1061,7 @@ record_dead_member_in_group(Member, GroupName) ->
Group. Group.
record_new_member_in_group(GroupName, Left, NewMember, Fun) -> record_new_member_in_group(GroupName, Left, NewMember, Fun) ->
{atomic, Group} = {atomic, {Result, Group}} =
mnesia:sync_transaction( mnesia:sync_transaction(
fun () -> fun () ->
[#gm_group { members = Members, version = Ver } = Group1] = [#gm_group { members = Members, version = Ver } = Group1] =
@ -1062,11 +1071,11 @@ record_new_member_in_group(GroupName, Left, NewMember, Fun) ->
Members1 = Prefix ++ [Left, NewMember | Suffix], Members1 = Prefix ++ [Left, NewMember | Suffix],
Group2 = Group1 #gm_group { members = Members1, Group2 = Group1 #gm_group { members = Members1,
version = Ver + 1 }, version = Ver + 1 },
ok = Fun(Group2), Result = Fun(Group2),
mnesia:write(Group2), mnesia:write(Group2),
Group2 {Result, Group2}
end), end),
Group. {Result, Group}.
erase_members_in_group(Members, GroupName) -> erase_members_in_group(Members, GroupName) ->
DeadMembers = [{dead, Id} || Id <- Members], DeadMembers = [{dead, Id} || Id <- Members],
@ -1089,10 +1098,10 @@ erase_members_in_group(Members, GroupName) ->
maybe_erase_aliases(State = #state { self = Self, maybe_erase_aliases(State = #state { self = Self,
group_name = GroupName, group_name = GroupName,
view = View, view = View0,
members_state = MembersState, members_state = MembersState,
module = Module, module = Module,
callback_args = Args }) -> callback_args = Args }, View) ->
#view_member { aliases = Aliases } = fetch_view_member(Self, View), #view_member { aliases = Aliases } = fetch_view_member(Self, View),
{Erasable, MembersState1} {Erasable, MembersState1}
= ?SETS:fold( = ?SETS:fold(
@ -1107,11 +1116,11 @@ maybe_erase_aliases(State = #state { self = Self,
end, {[], MembersState}, Aliases), end, {[], MembersState}, Aliases),
State1 = State #state { members_state = MembersState1 }, State1 = State #state { members_state = MembersState1 },
case Erasable of case Erasable of
[] -> {ok, State1}; [] -> {ok, State1 #state { view = View }};
_ -> View1 = group_to_view( _ -> View1 = group_to_view(
erase_members_in_group(Erasable, GroupName)), erase_members_in_group(Erasable, GroupName)),
{callback_view_changed(Args, Module, View, View1), {callback_view_changed(Args, Module, View0, View1),
State1 #state { view = View1 }} check_neighbours(State1 #state { view = View1 })}
end. end.
can_erase_view_member(Self, Self, _LA, _LP) -> false; can_erase_view_member(Self, Self, _LA, _LP) -> false;
@ -1257,6 +1266,12 @@ make_member(GroupName) ->
{error, not_found} -> ?VERSION_START {error, not_found} -> ?VERSION_START
end, self()}. end, self()}.
remove_erased_members(MembersState, View) ->
lists:foldl(fun (Id, MembersState1) ->
store_member(Id, find_member_or_blank(Id, MembersState),
MembersState1)
end, blank_member_state(), all_known_members(View)).
get_pid({_Version, Pid}) -> Pid. get_pid({_Version, Pid}) -> Pid.
get_pids(Ids) -> [Pid || {_Version, Pid} <- Ids]. get_pids(Ids) -> [Pid || {_Version, Pid} <- Ids].
@ -1287,16 +1302,30 @@ send_right(Right, View, Msg) ->
ok = gen_server2:cast(get_pid(Right), {?TAG, view_version(View), Msg}). ok = gen_server2:cast(get_pid(Right), {?TAG, view_version(View), Msg}).
callback(Args, Module, Activity) -> callback(Args, Module, Activity) ->
lists:foldl( Result =
fun ({Id, Pubs, _Acks}, ok) -> lists:foldl(
lists:foldl(fun ({_PubNum, Pub}, ok) -> fun ({Id, Pubs, _Acks}, {Args1, Module1, ok}) ->
Module:handle_msg(Args, get_pid(Id), Pub); lists:foldl(fun ({_PubNum, Pub}, Acc = {Args2, Module2, ok}) ->
(_, Error) -> case Module2:handle_msg(
Error Args2, get_pid(Id), Pub) of
end, ok, Pubs); ok ->
(_, Error) -> Acc;
Error {become, Module3, Args3} ->
end, ok, Activity). {Args3, Module3, ok};
{stop, _Reason} = Error ->
Error
end;
(_, Error = {stop, _Reason}) ->
Error
end, {Args1, Module1, ok}, Pubs);
(_, Error = {stop, _Reason}) ->
Error
end, {Args, Module, ok}, Activity),
case Result of
{Args, Module, ok} -> ok;
{Args1, Module1, ok} -> {become, Module1, Args1};
{stop, _Reason} = Error -> Error
end.
callback_view_changed(Args, Module, OldView, NewView) -> callback_view_changed(Args, Module, OldView, NewView) ->
OldMembers = all_known_members(OldView), OldMembers = all_known_members(OldView),

122
src/mirrored_supervisor.erl
View File

@ -225,8 +225,8 @@ which_children(Sup) -> fold(which_children, Sup, fun lists:append/2).
count_children(Sup) -> fold(count_children, Sup, fun add_proplists/2). count_children(Sup) -> fold(count_children, Sup, fun add_proplists/2).
check_childspecs(Specs) -> ?SUPERVISOR:check_childspecs(Specs). check_childspecs(Specs) -> ?SUPERVISOR:check_childspecs(Specs).
call(Sup, Msg) -> call(Sup, Msg) -> ?GEN_SERVER:call(mirroring(Sup), Msg, infinity).
?GEN_SERVER:call(child(Sup, mirroring), Msg, infinity). cast(Sup, Msg) -> ?GEN_SERVER:cast(mirroring(Sup), Msg).
find_call(Sup, Id, Msg) -> find_call(Sup, Id, Msg) ->
Group = call(Sup, group), Group = call(Sup, group),
@ -237,7 +237,7 @@ find_call(Sup, Id, Msg) ->
%% immediately after the tx - we can't be 100% here. So we may as %% immediately after the tx - we can't be 100% here. So we may as
%% well dirty_select. %% well dirty_select.
case mnesia:dirty_select(?TABLE, [{MatchHead, [], ['$1']}]) of case mnesia:dirty_select(?TABLE, [{MatchHead, [], ['$1']}]) of
[Mirror] -> ?GEN_SERVER:call(Mirror, Msg, infinity); [Mirror] -> call(Mirror, Msg);
[] -> {error, not_found} [] -> {error, not_found}
end. end.
@ -246,13 +246,16 @@ fold(FunAtom, Sup, AggFun) ->
lists:foldl(AggFun, [], lists:foldl(AggFun, [],
[apply(?SUPERVISOR, FunAtom, [D]) || [apply(?SUPERVISOR, FunAtom, [D]) ||
M <- ?PG2:get_members(Group), M <- ?PG2:get_members(Group),
D <- [?GEN_SERVER:call(M, delegate_supervisor, infinity)]]). D <- [delegate(M)]]).
child(Sup, Id) -> child(Sup, Id) ->
[Pid] = [Pid || {Id1, Pid, _, _} <- ?SUPERVISOR:which_children(Sup), [Pid] = [Pid || {Id1, Pid, _, _} <- ?SUPERVISOR:which_children(Sup),
Id1 =:= Id], Id1 =:= Id],
Pid. Pid.
delegate(Sup) -> child(Sup, delegate).
mirroring(Sup) -> child(Sup, mirroring).
%%---------------------------------------------------------------------------- %%----------------------------------------------------------------------------
start_internal(Group, ChildSpecs) -> start_internal(Group, ChildSpecs) ->
@ -261,24 +264,19 @@ start_internal(Group, ChildSpecs) ->
%%---------------------------------------------------------------------------- %%----------------------------------------------------------------------------
init({overall, Group, Init}) -> init({overall, _Group, ignore}) -> ignore;
case Init of init({overall, Group, {ok, {Restart, ChildSpecs}}}) ->
{ok, {Restart, ChildSpecs}} -> %% Important: Delegate MUST start before Mirroring so that when we
Delegate = {delegate, {?SUPERVISOR, start_link, %% shut down from above it shuts down last, so Mirroring does not
[?MODULE, {delegate, Restart}]}, %% see it die.
temporary, 16#ffffffff, supervisor, [?SUPERVISOR]}, %%
Mirroring = {mirroring, {?MODULE, start_internal, %% See comment in handle_info('DOWN', ...) below
[Group, ChildSpecs]}, {ok, {{one_for_all, 0, 1},
permanent, 16#ffffffff, worker, [?MODULE]}, [{delegate, {?SUPERVISOR, start_link, [?MODULE, {delegate, Restart}]},
%% Important: Delegate MUST start before Mirroring so that temporary, 16#ffffffff, supervisor, [?SUPERVISOR]},
%% when we shut down from above it shuts down last, so {mirroring, {?MODULE, start_internal, [Group, ChildSpecs]},
%% Mirroring does not see it die. permanent, 16#ffffffff, worker, [?MODULE]}]}};
%%
%% See comment in handle_info('DOWN', ...) below
{ok, {{one_for_all, 0, 1}, [Delegate, Mirroring]}};
ignore ->
ignore
end;
init({delegate, Restart}) -> init({delegate, Restart}) ->
{ok, {Restart, []}}; {ok, {Restart, []}};
@ -293,28 +291,29 @@ handle_call({init, Overall}, _From,
initial_childspecs = ChildSpecs}) -> initial_childspecs = ChildSpecs}) ->
process_flag(trap_exit, true), process_flag(trap_exit, true),
?PG2:create(Group), ?PG2:create(Group),
ok = ?PG2:join(Group, self()), ok = ?PG2:join(Group, Overall),
Rest = ?PG2:get_members(Group) -- [self()], Rest = ?PG2:get_members(Group) -- [Overall],
case Rest of case Rest of
[] -> {atomic, _} = mnesia:transaction(fun() -> delete_all(Group) end); [] -> {atomic, _} = mnesia:transaction(fun() -> delete_all(Group) end);
_ -> ok _ -> ok
end, end,
[begin [begin
?GEN_SERVER:cast(Pid, {ensure_monitoring, self()}), ?GEN_SERVER:cast(mirroring(Pid), {ensure_monitoring, Overall}),
erlang:monitor(process, Pid) erlang:monitor(process, Pid)
end || Pid <- Rest], end || Pid <- Rest],
Delegate = child(Overall, delegate), Delegate = delegate(Overall),
erlang:monitor(process, Delegate), erlang:monitor(process, Delegate),
State1 = State#state{overall = Overall, delegate = Delegate}, State1 = State#state{overall = Overall, delegate = Delegate},
case errors([maybe_start(Group, Delegate, S) || S <- ChildSpecs]) of case errors([maybe_start(Group, Overall, Delegate, S) || S <- ChildSpecs]) of
[] -> {reply, ok, State1}; [] -> {reply, ok, State1};
Errors -> {stop, {shutdown, {init, Errors, ChildSpecs}}, State1} Errors -> {stop, {shutdown, Errors}, State1}
end; end;
handle_call({start_child, ChildSpec}, _From, handle_call({start_child, ChildSpec}, _From,
State = #state{delegate = Delegate, State = #state{overall = Overall,
delegate = Delegate,
group = Group}) -> group = Group}) ->
{reply, case maybe_start(Group, Delegate, ChildSpec) of {reply, case maybe_start(Group, Overall, Delegate, ChildSpec) of
already_in_mnesia -> {error, already_present}; already_in_mnesia -> {error, already_present};
{already_in_mnesia, Pid} -> {error, {already_started, Pid}}; {already_in_mnesia, Pid} -> {error, {already_started, Pid}};
Else -> Else Else -> Else
@ -327,9 +326,6 @@ handle_call({delete_child, Id}, _From, State = #state{delegate = Delegate,
handle_call({msg, F, A}, _From, State = #state{delegate = Delegate}) -> handle_call({msg, F, A}, _From, State = #state{delegate = Delegate}) ->
{reply, apply(?SUPERVISOR, F, [Delegate | A]), State}; {reply, apply(?SUPERVISOR, F, [Delegate | A]), State};
handle_call(delegate_supervisor, _From, State = #state{delegate = Delegate}) ->
{reply, Delegate, State};
handle_call(group, _From, State = #state{group = Group}) -> handle_call(group, _From, State = #state{group = Group}) ->
{reply, Group, State}; {reply, Group, State};
@ -348,7 +344,7 @@ handle_cast(Msg, State) ->
{stop, {unexpected_cast, Msg}, State}. {stop, {unexpected_cast, Msg}, State}.
handle_info({'DOWN', _Ref, process, Pid, Reason}, handle_info({'DOWN', _Ref, process, Pid, Reason},
State = #state{delegate = Pid, group = Group}) -> State = #state{overall = Pid, group = Group}) ->
%% Since the delegate is temporary, its death won't cause us to %% Since the delegate is temporary, its death won't cause us to
%% die. Since the overall supervisor kills processes in reverse %% die. Since the overall supervisor kills processes in reverse
%% order when shutting down "from above" and we started after the %% order when shutting down "from above" and we started after the
@ -362,21 +358,20 @@ handle_info({'DOWN', _Ref, process, Pid, Reason},
{stop, Reason, State}; {stop, Reason, State};
handle_info({'DOWN', _Ref, process, Pid, _Reason}, handle_info({'DOWN', _Ref, process, Pid, _Reason},
State = #state{delegate = Delegate, group = Group}) -> State = #state{delegate = Delegate, group = Group,
overall = O}) ->
%% TODO load balance this %% TODO load balance this
%% No guarantee pg2 will have received the DOWN before us. %% No guarantee pg2 will have received the DOWN before us.
Self = self(), R = case lists:sort(?PG2:get_members(Group)) -- [Pid] of
{R, Cs, X} = [O | _] -> {atomic, ChildSpecs} =
case lists:sort(?PG2:get_members(Group)) -- [Pid] of mnesia:transaction(
[Self | _] -> {atomic, {ChildSpecs, Extra}} = fun() -> update_all(O, Pid) end),
mnesia:transaction(fun() -> update_all(Pid) end), [start(Delegate, ChildSpec) || ChildSpec <- ChildSpecs];
{[start(Delegate, ChildSpec) || ChildSpec <- ChildSpecs], _ -> []
ChildSpecs, Extra};
_ -> {[], [], []}
end, end,
case errors(R) of case errors(R) of
[] -> {noreply, State}; [] -> {noreply, State};
Errors -> {stop, {shutdown, {down, Errors, Cs, X}}, State} Errors -> {stop, {shutdown, Errors}, State}
end; end;
handle_info(Info, State) -> handle_info(Info, State) ->
@ -391,13 +386,11 @@ code_change(_OldVsn, State, _Extra) ->
%%---------------------------------------------------------------------------- %%----------------------------------------------------------------------------
tell_all_peers_to_die(Group, Reason) -> tell_all_peers_to_die(Group, Reason) ->
[?GEN_SERVER:cast(P, {die, Reason}) || [cast(P, {die, Reason}) || P <- ?PG2:get_members(Group) -- [self()]].
P <- ?PG2:get_members(Group) -- [self()]].
maybe_start(Group, Delegate, ChildSpec) -> maybe_start(Group, Overall, Delegate, ChildSpec) ->
case mnesia:transaction(fun() -> case mnesia:transaction(
check_start(Group, Delegate, ChildSpec) fun() -> check_start(Group, Overall, Delegate, ChildSpec) end) of
end) of
{atomic, start} -> start(Delegate, ChildSpec); {atomic, start} -> start(Delegate, ChildSpec);
{atomic, undefined} -> already_in_mnesia; {atomic, undefined} -> already_in_mnesia;
{atomic, Pid} -> {already_in_mnesia, Pid}; {atomic, Pid} -> {already_in_mnesia, Pid};
@ -405,31 +398,29 @@ maybe_start(Group, Delegate, ChildSpec) ->
{aborted, E} -> {error, E} {aborted, E} -> {error, E}
end. end.
check_start(Group, Delegate, ChildSpec) -> check_start(Group, Overall, Delegate, ChildSpec) ->
case mnesia:wread({?TABLE, {Group, id(ChildSpec)}}) of case mnesia:wread({?TABLE, {Group, id(ChildSpec)}}) of
[] -> write(Group, ChildSpec), [] -> write(Group, Overall, ChildSpec),
start; start;
[S] -> #mirrored_sup_childspec{key = {Group, Id}, [S] -> #mirrored_sup_childspec{key = {Group, Id},
mirroring_pid = Pid} = S, mirroring_pid = Pid} = S,
case self() of case Overall of
Pid -> child(Delegate, Id); Pid -> child(Delegate, Id);
_ -> case supervisor(Pid) of _ -> case supervisor(Pid) of
dead -> write(Group, ChildSpec), dead -> write(Group, Overall, ChildSpec),
start; start;
Delegate0 -> child(Delegate0, Id) Delegate0 -> child(Delegate0, Id)
end end
end end
end. end.
supervisor(Pid) -> supervisor(Pid) -> with_exit_handler(fun() -> dead end,
with_exit_handler( fun() -> delegate(Pid) end).
fun() -> dead end,
fun() -> gen_server:call(Pid, delegate_supervisor, infinity) end).
write(Group, ChildSpec) -> write(Group, Overall, ChildSpec) ->
ok = mnesia:write( ok = mnesia:write(
#mirrored_sup_childspec{key = {Group, id(ChildSpec)}, #mirrored_sup_childspec{key = {Group, id(ChildSpec)},
mirroring_pid = self(), mirroring_pid = Overall,
childspec = ChildSpec}), childspec = ChildSpec}),
ChildSpec. ChildSpec.
@ -455,13 +446,13 @@ check_stop(Group, Delegate, Id) ->
id({Id, _, _, _, _, _}) -> Id. id({Id, _, _, _, _, _}) -> Id.
update_all(OldPid) -> update_all(Overall, OldOverall) ->
MatchHead = #mirrored_sup_childspec{mirroring_pid = OldPid, MatchHead = #mirrored_sup_childspec{mirroring_pid = OldOverall,
key = '$1', key = '$1',
childspec = '$2', childspec = '$2',
_ = '_'}, _ = '_'},
Matches = mnesia:select(?TABLE, [{MatchHead, [], ['$$']}]), [write(Group, Overall, C) ||
{[write(Group, C) || [{Group, _Id}, C] <- Matches], {OldPid, Matches}}. [{Group, _Id}, C] <- mnesia:select(?TABLE, [{MatchHead, [], ['$$']}])].
delete_all(Group) -> delete_all(Group) ->
MatchHead = #mirrored_sup_childspec{key = {Group, '_'}, MatchHead = #mirrored_sup_childspec{key = {Group, '_'},
@ -474,8 +465,7 @@ errors(Results) -> [E || {error, E} <- Results].
%%---------------------------------------------------------------------------- %%----------------------------------------------------------------------------
create_tables() -> create_tables() -> create_tables([?TABLE_DEF]).
create_tables([?TABLE_DEF]).
create_tables([]) -> create_tables([]) ->
ok; ok;

24
src/mirrored_supervisor_tests.erl
View File

@ -51,7 +51,7 @@ test_migrate() ->
with_sups(fun([A, _]) -> with_sups(fun([A, _]) ->
?MS:start_child(a, childspec(worker)), ?MS:start_child(a, childspec(worker)),
Pid1 = pid_of(worker), Pid1 = pid_of(worker),
kill(A, Pid1), kill_registered(A, Pid1),
Pid2 = pid_of(worker), Pid2 = pid_of(worker),
false = (Pid1 =:= Pid2) false = (Pid1 =:= Pid2)
end, [a, b]). end, [a, b]).
@ -61,10 +61,10 @@ test_migrate_twice() ->
with_sups(fun([A, B]) -> with_sups(fun([A, B]) ->
?MS:start_child(a, childspec(worker)), ?MS:start_child(a, childspec(worker)),
Pid1 = pid_of(worker), Pid1 = pid_of(worker),
kill(A, Pid1), kill_registered(A, Pid1),
{ok, C} = start_sup(c), {ok, C} = start_sup(c),
Pid2 = pid_of(worker), Pid2 = pid_of(worker),
kill(B, Pid2), kill_registered(B, Pid2),
Pid3 = pid_of(worker), Pid3 = pid_of(worker),
false = (Pid1 =:= Pid3), false = (Pid1 =:= Pid3),
kill(C) kill(C)
@ -124,7 +124,7 @@ test_large_group() ->
with_sups(fun([A, _, _, _]) -> with_sups(fun([A, _, _, _]) ->
?MS:start_child(a, childspec(worker)), ?MS:start_child(a, childspec(worker)),
Pid1 = pid_of(worker), Pid1 = pid_of(worker),
kill(A, Pid1), kill_registered(A, Pid1),
Pid2 = pid_of(worker), Pid2 = pid_of(worker),
false = (Pid1 =:= Pid2) false = (Pid1 =:= Pid2)
end, [a, b, c, d]). end, [a, b, c, d]).
@ -134,7 +134,7 @@ test_childspecs_at_init() ->
S = childspec(worker), S = childspec(worker),
with_sups(fun([A, _]) -> with_sups(fun([A, _]) ->
Pid1 = pid_of(worker), Pid1 = pid_of(worker),
kill(A, Pid1), kill_registered(A, Pid1),
Pid2 = pid_of(worker), Pid2 = pid_of(worker),
false = (Pid1 =:= Pid2) false = (Pid1 =:= Pid2)
end, [{a, [S]}, {b, [S]}]). end, [{a, [S]}, {b, [S]}]).
@ -143,7 +143,7 @@ test_anonymous_supervisors() ->
with_sups(fun([A, _B]) -> with_sups(fun([A, _B]) ->
?MS:start_child(A, childspec(worker)), ?MS:start_child(A, childspec(worker)),
Pid1 = pid_of(worker), Pid1 = pid_of(worker),
kill(A, Pid1), kill_registered(A, Pid1),
Pid2 = pid_of(worker), Pid2 = pid_of(worker),
false = (Pid1 =:= Pid2) false = (Pid1 =:= Pid2)
end, [anon, anon]). end, [anon, anon]).
@ -157,7 +157,7 @@ test_no_migration_on_shutdown() ->
with_sups(fun([Evil, _]) -> with_sups(fun([Evil, _]) ->
?MS:start_child(Evil, childspec(worker)), ?MS:start_child(Evil, childspec(worker)),
try try
call(worker, ping, 10000, 100), call(worker, ping, 1000, 100),
exit(worker_should_not_have_migrated) exit(worker_should_not_have_migrated)
catch exit:{timeout_waiting_for_server, _, _} -> catch exit:{timeout_waiting_for_server, _, _} ->
ok ok
@ -268,7 +268,7 @@ inc_group() ->
get_group(Group) -> get_group(Group) ->
{Group, get(counter)}. {Group, get(counter)}.
call(Id, Msg) -> call(Id, Msg, 5*24*60*60*1000, 100). call(Id, Msg) -> call(Id, Msg, 10*1000, 100).
call(Id, Msg, 0, _Decr) -> call(Id, Msg, 0, _Decr) ->
exit({timeout_waiting_for_server, {Id, Msg}, erlang:get_stacktrace()}); exit({timeout_waiting_for_server, {Id, Msg}, erlang:get_stacktrace()});
@ -285,10 +285,16 @@ kill(Pid, Wait) when is_pid(Wait) -> kill(Pid, [Wait]);
kill(Pid, Waits) -> kill(Pid, Waits) ->
erlang:monitor(process, Pid), erlang:monitor(process, Pid),
[erlang:monitor(process, P) || P <- Waits], [erlang:monitor(process, P) || P <- Waits],
exit(Pid, kill), exit(Pid, bang),
kill_wait(Pid), kill_wait(Pid),
[kill_wait(P) || P <- Waits]. [kill_wait(P) || P <- Waits].
kill_registered(Pid, Child) ->
{registered_name, Name} = erlang:process_info(Child, registered_name),
kill(Pid, Child),
false = (Child =:= whereis(Name)),
ok.
kill_wait(Pid) -> kill_wait(Pid) ->
receive receive
{'DOWN', _Ref, process, Pid, _Reason} -> {'DOWN', _Ref, process, Pid, _Reason} ->

157
src/rabbit.erl
View File

@ -18,9 +18,9 @@
-behaviour(application). -behaviour(application).
-export([maybe_hipe_compile/0, prepare/0, start/0, stop/0, stop_and_halt/0, -export([start/0, boot/0, stop/0,
status/0, is_running/0, is_running/1, environment/0, stop_and_halt/0, await_startup/0, status/0, is_running/0,
rotate_logs/1, force_event_refresh/0]). is_running/1, environment/0, rotate_logs/1, force_event_refresh/0]).
-export([start/2, stop/1]). -export([start/2, stop/1]).
@ -60,7 +60,8 @@
-rabbit_boot_step({worker_pool, -rabbit_boot_step({worker_pool,
[{description, "worker pool"}, [{description, "worker pool"},
{mfa, {rabbit_sup, start_child, [worker_pool_sup]}}, {mfa, {rabbit_sup, start_supervisor_child,
[worker_pool_sup]}},
{requires, pre_boot}, {requires, pre_boot},
{enables, external_infrastructure}]}). {enables, external_infrastructure}]}).
@ -143,7 +144,8 @@
-rabbit_boot_step({mirror_queue_slave_sup, -rabbit_boot_step({mirror_queue_slave_sup,
[{description, "mirror queue slave sup"}, [{description, "mirror queue slave sup"},
{mfa, {rabbit_mirror_queue_slave_sup, start, []}}, {mfa, {rabbit_sup, start_supervisor_child,
[rabbit_mirror_queue_slave_sup]}},
{requires, recovery}, {requires, recovery},
{enables, routing_ready}]}). {enables, routing_ready}]}).
@ -197,7 +199,7 @@
rabbit_queue_index, gen, dict, ordsets, file_handle_cache, rabbit_queue_index, gen, dict, ordsets, file_handle_cache,
rabbit_msg_store, array, rabbit_msg_store_ets_index, rabbit_msg_file, rabbit_msg_store, array, rabbit_msg_store_ets_index, rabbit_msg_file,
rabbit_exchange_type_fanout, rabbit_exchange_type_topic, mnesia, rabbit_exchange_type_fanout, rabbit_exchange_type_topic, mnesia,
mnesia_lib, rpc, mnesia_tm, qlc, sofs, proplists, credit_flow]). mnesia_lib, rpc, mnesia_tm, qlc, sofs, proplists, credit_flow, pmon]).
%% HiPE compilation uses multiple cores anyway, but some bits are %% HiPE compilation uses multiple cores anyway, but some bits are
%% IO-bound so we can go faster if we parallelise a bit more. In %% IO-bound so we can go faster if we parallelise a bit more. In
@ -214,11 +216,11 @@
-type(log_location() :: 'tty' | 'undefined' | file:filename()). -type(log_location() :: 'tty' | 'undefined' | file:filename()).
-type(param() :: atom()). -type(param() :: atom()).
-spec(maybe_hipe_compile/0 :: () -> 'ok').
-spec(prepare/0 :: () -> 'ok').
-spec(start/0 :: () -> 'ok'). -spec(start/0 :: () -> 'ok').
-spec(boot/0 :: () -> 'ok').
-spec(stop/0 :: () -> 'ok'). -spec(stop/0 :: () -> 'ok').
-spec(stop_and_halt/0 :: () -> no_return()). -spec(stop_and_halt/0 :: () -> no_return()).
-spec(await_startup/0 :: () -> 'ok').
-spec(status/0 :: -spec(status/0 ::
() -> [{pid, integer()} | () -> [{pid, integer()} |
{running_applications, [{atom(), string(), string()}]} | {running_applications, [{atom(), string(), string()}]} |
@ -283,29 +285,51 @@ split(L, N) -> split0(L, [[] || _ <- lists:seq(1, N)]).
split0([], Ls) -> Ls; split0([], Ls) -> Ls;
split0([I | Is], [L | Ls]) -> split0(Is, Ls ++ [[I | L]]). split0([I | Is], [L | Ls]) -> split0(Is, Ls ++ [[I | L]]).
prepare() -> ensure_application_loaded() ->
ok = ensure_working_log_handlers(), %% We end up looking at the rabbit app's env for HiPE and log
ok = rabbit_upgrade:maybe_upgrade_mnesia(). %% handling, so it needs to be loaded. But during the tests, it
%% may end up getting loaded twice, so guard against that.
case application:load(rabbit) of
ok -> ok;
{error, {already_loaded, rabbit}} -> ok
end.
start() -> start() ->
start_it(fun() ->
%% We do not want to HiPE compile or upgrade
%% mnesia after just restarting the app
ok = ensure_application_loaded(),
ok = ensure_working_log_handlers(),
ok = app_utils:start_applications(app_startup_order()),
ok = print_plugin_info(rabbit_plugins:active())
end).
boot() ->
start_it(fun() ->
ok = ensure_application_loaded(),
maybe_hipe_compile(),
ok = ensure_working_log_handlers(),
ok = rabbit_upgrade:maybe_upgrade_mnesia(),
Plugins = rabbit_plugins:setup(),
ToBeLoaded = Plugins ++ ?APPS,
ok = app_utils:load_applications(ToBeLoaded),
StartupApps = app_utils:app_dependency_order(ToBeLoaded,
false),
ok = app_utils:start_applications(StartupApps),
ok = print_plugin_info(Plugins)
end).
start_it(StartFun) ->
try try
%% prepare/1 ends up looking at the rabbit app's env, so it StartFun()
%% needs to be loaded, but during the tests, it may end up
%% getting loaded twice, so guard against that
case application:load(rabbit) of
ok -> ok;
{error, {already_loaded, rabbit}} -> ok
end,
ok = prepare(),
ok = rabbit_misc:start_applications(application_load_order())
after after
%%give the error loggers some time to catch up %% give the error loggers some time to catch up
timer:sleep(100) timer:sleep(100)
end. end.
stop() -> stop() ->
rabbit_log:info("Stopping Rabbit~n"), rabbit_log:info("Stopping Rabbit~n"),
ok = rabbit_misc:stop_applications(application_load_order()). ok = app_utils:stop_applications(app_shutdown_order()).
stop_and_halt() -> stop_and_halt() ->
try try
@ -316,6 +340,9 @@ stop_and_halt() ->
end, end,
ok. ok.
await_startup() ->
app_utils:wait_for_applications(app_startup_order()).
status() -> status() ->
S1 = [{pid, list_to_integer(os:getpid())}, S1 = [{pid, list_to_integer(os:getpid())},
{running_applications, application:which_applications(infinity)}, {running_applications, application:which_applications(infinity)},
@ -392,46 +419,13 @@ stop(_State) ->
%%--------------------------------------------------------------------------- %%---------------------------------------------------------------------------
%% application life cycle %% application life cycle
application_load_order() -> app_startup_order() ->
ok = load_applications(), ok = app_utils:load_applications(?APPS),
{ok, G} = rabbit_misc:build_acyclic_graph( app_utils:app_dependency_order(?APPS, false).
fun (App, _Deps) -> [{App, App}] end,
fun (App, Deps) -> [{Dep, App} || Dep <- Deps] end,
[{App, app_dependencies(App)} ||
{App, _Desc, _Vsn} <- application:loaded_applications()]),
true = digraph:del_vertices(
G, digraph:vertices(G) -- digraph_utils:reachable(?APPS, G)),
Result = digraph_utils:topsort(G),
true = digraph:delete(G),
Result.
load_applications() -> app_shutdown_order() ->
load_applications(queue:from_list(?APPS), sets:new()). Apps = ?APPS ++ rabbit_plugins:active(),
app_utils:app_dependency_order(Apps, true).
load_applications(Worklist, Loaded) ->
case queue:out(Worklist) of
{empty, _WorkList} ->
ok;
{{value, App}, Worklist1} ->
case sets:is_element(App, Loaded) of
true -> load_applications(Worklist1, Loaded);
false -> case application:load(App) of
ok -> ok;
{error, {already_loaded, App}} -> ok;
Error -> throw(Error)
end,
load_applications(
queue:join(Worklist1,
queue:from_list(app_dependencies(App))),
sets:add_element(App, Loaded))
end
end.
app_dependencies(App) ->
case application:get_key(App, applications) of
undefined -> [];
{ok, Lst} -> Lst
end.
%%--------------------------------------------------------------------------- %%---------------------------------------------------------------------------
%% boot step logic %% boot step logic
@ -477,7 +471,8 @@ sort_boot_steps(UnsortedSteps) ->
%% there is one, otherwise fail). %% there is one, otherwise fail).
SortedSteps = lists:reverse( SortedSteps = lists:reverse(
[begin [begin
{StepName, Step} = digraph:vertex(G, StepName), {StepName, Step} = digraph:vertex(G,
StepName),
Step Step
end || StepName <- digraph_utils:topsort(G)]), end || StepName <- digraph_utils:topsort(G)]),
digraph:delete(G), digraph:delete(G),
@ -538,7 +533,7 @@ boot_error(Format, Args) ->
boot_delegate() -> boot_delegate() ->
{ok, Count} = application:get_env(rabbit, delegate_count), {ok, Count} = application:get_env(rabbit, delegate_count),
rabbit_sup:start_child(delegate_sup, [Count]). rabbit_sup:start_supervisor_child(delegate_sup, [Count]).
recover() -> recover() ->
rabbit_binding:recover(rabbit_exchange:recover(), rabbit_amqqueue:start()). rabbit_binding:recover(rabbit_exchange:recover(), rabbit_amqqueue:start()).
@ -559,7 +554,8 @@ insert_default_data() ->
ok = rabbit_vhost:add(DefaultVHost), ok = rabbit_vhost:add(DefaultVHost),
ok = rabbit_auth_backend_internal:add_user(DefaultUser, DefaultPass), ok = rabbit_auth_backend_internal:add_user(DefaultUser, DefaultPass),
ok = rabbit_auth_backend_internal:set_tags(DefaultUser, DefaultTags), ok = rabbit_auth_backend_internal:set_tags(DefaultUser, DefaultTags),
ok = rabbit_auth_backend_internal:set_permissions(DefaultUser, DefaultVHost, ok = rabbit_auth_backend_internal:set_permissions(DefaultUser,
DefaultVHost,
DefaultConfigurePerm, DefaultConfigurePerm,
DefaultWritePerm, DefaultWritePerm,
DefaultReadPerm), DefaultReadPerm),
@ -621,15 +617,12 @@ log_location(Type) ->
rotate_logs(File, Suffix, Handler) -> rotate_logs(File, Suffix, Handler) ->
rotate_logs(File, Suffix, Handler, Handler). rotate_logs(File, Suffix, Handler, Handler).
rotate_logs(File, Suffix, OldHandler, NewHandler) -> rotate_logs(undefined, _Suffix, _OldHandler, _NewHandler) -> ok;
case File of rotate_logs(tty, _Suffix, _OldHandler, _NewHandler) -> ok;
undefined -> ok; rotate_logs(File, Suffix, OldHandler, NewHandler) ->
tty -> ok; gen_event:swap_handler(error_logger,
_ -> gen_event:swap_handler( {OldHandler, swap},
error_logger, {NewHandler, {File, Suffix}}).
{OldHandler, swap},
{NewHandler, {File, Suffix}})
end.
log_rotation_result({error, MainLogError}, {error, SaslLogError}) -> log_rotation_result({error, MainLogError}, {error, SaslLogError}) ->
{error, {{cannot_rotate_main_logs, MainLogError}, {error, {{cannot_rotate_main_logs, MainLogError},
@ -650,6 +643,24 @@ force_event_refresh() ->
%%--------------------------------------------------------------------------- %%---------------------------------------------------------------------------
%% misc %% misc
print_plugin_info([]) ->
ok;
print_plugin_info(Plugins) ->
%% This gets invoked by rabbitmqctl start_app, outside the context
%% of the rabbit application
rabbit_misc:with_local_io(
fun() ->
io:format("~n-- plugins running~n"),
[print_plugin_info(
AppName, element(2, application:get_key(AppName, vsn)))
|| AppName <- Plugins],
ok
end).
print_plugin_info(Plugin, Vsn) ->
Len = 76 - length(Vsn),
io:format("~-" ++ integer_to_list(Len) ++ "s ~s~n", [Plugin, Vsn]).
erts_version_check() -> erts_version_check() ->
FoundVer = erlang:system_info(version), FoundVer = erlang:system_info(version),
case rabbit_misc:version_compare(?ERTS_MINIMUM, FoundVer, lte) of case rabbit_misc:version_compare(?ERTS_MINIMUM, FoundVer, lte) of

12
src/rabbit_alarm.erl
View File

@ -162,17 +162,17 @@ maybe_alert(UpdateFun, Node, Source,
end, end,
State#alarms{alarmed_nodes = AN1}. State#alarms{alarmed_nodes = AN1}.
alert_local(Alert, Alertees, _Source) -> alert_local(Alert, Alertees, Source) ->
alert(Alertees, [Alert], fun erlang:'=:='/2). alert(Alertees, Source, Alert, fun erlang:'=:='/2).
alert_remote(Alert, Alertees, Source) -> alert_remote(Alert, Alertees, Source) ->
alert(Alertees, [Source, Alert], fun erlang:'=/='/2). alert(Alertees, Source, Alert, fun erlang:'=/='/2).
alert(Alertees, AlertArg, NodeComparator) -> alert(Alertees, Source, Alert, NodeComparator) ->
Node = node(), Node = node(),
dict:fold(fun (Pid, {M, F, A}, ok) -> dict:fold(fun (Pid, {M, F, A}, ok) ->
case NodeComparator(Node, node(Pid)) of case NodeComparator(Node, node(Pid)) of
true -> apply(M, F, A ++ [Pid] ++ AlertArg); true -> apply(M, F, A ++ [Pid, Source, Alert]);
false -> ok false -> ok
end end
end, ok, Alertees). end, ok, Alertees).
@ -181,7 +181,7 @@ internal_register(Pid, {M, F, A} = HighMemMFA,
State = #alarms{alertees = Alertees}) -> State = #alarms{alertees = Alertees}) ->
_MRef = erlang:monitor(process, Pid), _MRef = erlang:monitor(process, Pid),
case dict:find(node(), State#alarms.alarmed_nodes) of case dict:find(node(), State#alarms.alarmed_nodes) of
{ok, _Sources} -> apply(M, F, A ++ [Pid, true]); {ok, Sources} -> [apply(M, F, A ++ [Pid, R, true]) || R <- Sources];
error -> ok error -> ok
end, end,
NewAlertees = dict:store(Pid, HighMemMFA, Alertees), NewAlertees = dict:store(Pid, HighMemMFA, Alertees),

6
src/rabbit_amqqueue_process.erl
View File

@ -984,8 +984,6 @@ prioritise_call(Msg, _From, _State) ->
info -> 9; info -> 9;
{info, _Items} -> 9; {info, _Items} -> 9;
consumers -> 9; consumers -> 9;
{basic_consume, _, _, _, _, _, _} -> 7;
{basic_cancel, _, _, _} -> 7;
stat -> 7; stat -> 7;
_ -> 0 _ -> 0
end. end.
@ -995,10 +993,6 @@ prioritise_cast(Msg, _State) ->
delete_immediately -> 8; delete_immediately -> 8;
{set_ram_duration_target, _Duration} -> 8; {set_ram_duration_target, _Duration} -> 8;
{set_maximum_since_use, _Age} -> 8; {set_maximum_since_use, _Age} -> 8;
{ack, _AckTags, _ChPid} -> 7;
{reject, _AckTags, _Requeue, _ChPid} -> 7;
{notify_sent, _ChPid, _Credit} -> 7;
{unblock, _ChPid} -> 7;
{run_backing_queue, _Mod, _Fun} -> 6; {run_backing_queue, _Mod, _Fun} -> 6;
_ -> 0 _ -> 0
end. end.

2
src/rabbit_backing_queue.erl
View File

@ -26,10 +26,8 @@
('empty' | ('empty' |
%% Message, IsDelivered, AckTag, Remaining_Len %% Message, IsDelivered, AckTag, Remaining_Len
{rabbit_types:basic_message(), boolean(), Ack, non_neg_integer()})). {rabbit_types:basic_message(), boolean(), Ack, non_neg_integer()})).
-type(is_durable() :: boolean()).
-type(attempt_recovery() :: boolean()). -type(attempt_recovery() :: boolean()).
-type(purged_msg_count() :: non_neg_integer()). -type(purged_msg_count() :: non_neg_integer()).
-type(confirm_required() :: boolean()).
-type(async_callback() :: fun ((atom(), fun ((atom(), state()) -> state())) -> 'ok')). -type(async_callback() :: fun ((atom(), fun ((atom(), state()) -> state())) -> 'ok')).
-type(duration() :: ('undefined' | 'infinity' | number())). -type(duration() :: ('undefined' | 'infinity' | number())).

3
src/rabbit_basic.erl
View File

@ -224,6 +224,5 @@ header_routes(HeadersTable) ->
{array, Routes} -> [Route || {longstr, Route} <- Routes]; {array, Routes} -> [Route || {longstr, Route} <- Routes];
undefined -> []; undefined -> [];
{Type, _Val} -> throw({error, {unacceptable_type_in_header, {Type, _Val} -> throw({error, {unacceptable_type_in_header,
Type, binary_to_list(HeaderKey), Type}})
binary_to_list(HeaderKey)}})
end || HeaderKey <- ?ROUTING_HEADERS]). end || HeaderKey <- ?ROUTING_HEADERS]).

121
src/rabbit_control.erl → src/rabbit_control_main.erl
View File

@ -14,7 +14,7 @@
%% Copyright (c) 2007-2012 VMware, Inc. All rights reserved. %% Copyright (c) 2007-2012 VMware, Inc. All rights reserved.
%% %%
-module(rabbit_control). -module(rabbit_control_main).
-include("rabbit.hrl"). -include("rabbit.hrl").
-export([start/0, stop/0, action/5]). -export([start/0, stop/0, action/5]).
@ -26,6 +26,61 @@
-define(NODE_OPT, "-n"). -define(NODE_OPT, "-n").
-define(VHOST_OPT, "-p"). -define(VHOST_OPT, "-p").
-define(QUIET_DEF, {?QUIET_OPT, flag}).
-define(NODE_DEF(Node), {?NODE_OPT, {option, Node}}).
-define(VHOST_DEF, {?VHOST_OPT, {option, "/"}}).
-define(GLOBAL_DEFS(Node), [?QUIET_DEF, ?NODE_DEF(Node)]).
-define(COMMANDS,
[stop,
stop_app,
start_app,
wait,
reset,
force_reset,
rotate_logs,
cluster,
force_cluster,
cluster_status,
add_user,
delete_user,
change_password,
clear_password,
set_user_tags,
list_users,
add_vhost,
delete_vhost,
list_vhosts,
{set_permissions, [?VHOST_DEF]},
{clear_permissions, [?VHOST_DEF]},
{list_permissions, [?VHOST_DEF]},
list_user_permissions,
set_parameter,
clear_parameter,
list_parameters,
{list_queues, [?VHOST_DEF]},
{list_exchanges, [?VHOST_DEF]},
{list_bindings, [?VHOST_DEF]},
{list_connections, [?VHOST_DEF]},
list_channels,
{list_consumers, [?VHOST_DEF]},
status,
environment,
report,
eval,
close_connection,
{trace_on, [?VHOST_DEF]},
{trace_off, [?VHOST_DEF]},
set_vm_memory_high_watermark
]).
-define(GLOBAL_QUERIES, -define(GLOBAL_QUERIES,
[{"Connections", rabbit_networking, connection_info_all, [{"Connections", rabbit_networking, connection_info_all,
connection_info_keys}, connection_info_keys},
@ -57,19 +112,18 @@
start() -> start() ->
{ok, [[NodeStr|_]|_]} = init:get_argument(nodename), {ok, [[NodeStr|_]|_]} = init:get_argument(nodename),
{[Command0 | Args], Opts} = {Command, Opts, Args} =
case rabbit_misc:get_options([{flag, ?QUIET_OPT}, case rabbit_misc:parse_arguments(?COMMANDS, ?GLOBAL_DEFS(NodeStr),
{option, ?NODE_OPT, NodeStr}, init:get_plain_arguments())
{option, ?VHOST_OPT, "/"}], of
init:get_plain_arguments()) of {ok, Res} -> Res;
{[], _Opts} -> usage(); no_command -> print_error("could not recognise command", []),
CmdArgsAndOpts -> CmdArgsAndOpts usage()
end, end,
Opts1 = [case K of Opts1 = [case K of
?NODE_OPT -> {?NODE_OPT, rabbit_nodes:make(V)}; ?NODE_OPT -> {?NODE_OPT, rabbit_nodes:make(V)};
_ -> {K, V} _ -> {K, V}
end || {K, V} <- Opts], end || {K, V} <- Opts],
Command = list_to_atom(Command0),
Quiet = proplists:get_bool(?QUIET_OPT, Opts1), Quiet = proplists:get_bool(?QUIET_OPT, Opts1),
Node = proplists:get_value(?NODE_OPT, Opts1), Node = proplists:get_value(?NODE_OPT, Opts1),
Inform = case Quiet of Inform = case Quiet of
@ -194,8 +248,7 @@ action(force_cluster, Node, ClusterNodeSs, _Opts, Inform) ->
action(wait, Node, [PidFile], _Opts, Inform) -> action(wait, Node, [PidFile], _Opts, Inform) ->
Inform("Waiting for ~p", [Node]), Inform("Waiting for ~p", [Node]),
wait_for_application(Node, PidFile, rabbit, Inform); wait_for_application(Node, PidFile, rabbit_and_plugins, Inform);
action(wait, Node, [PidFile, App], _Opts, Inform) -> action(wait, Node, [PidFile, App], _Opts, Inform) ->
Inform("Waiting for ~p on ~p", [App, Node]), Inform("Waiting for ~p on ~p", [App, Node]),
wait_for_application(Node, PidFile, list_to_atom(App), Inform); wait_for_application(Node, PidFile, list_to_atom(App), Inform);
@ -216,33 +269,33 @@ action(rotate_logs, Node, [], _Opts, Inform) ->
Inform("Reopening logs for node ~p", [Node]), Inform("Reopening logs for node ~p", [Node]),
call(Node, {rabbit, rotate_logs, [""]}); call(Node, {rabbit, rotate_logs, [""]});
action(rotate_logs, Node, Args = [Suffix], _Opts, Inform) -> action(rotate_logs, Node, Args = [Suffix], _Opts, Inform) ->
Inform("Rotating logs to files with suffix ~p", [Suffix]), Inform("Rotating logs to files with suffix \"~s\"", [Suffix]),
call(Node, {rabbit, rotate_logs, Args}); call(Node, {rabbit, rotate_logs, Args});
action(close_connection, Node, [PidStr, Explanation], _Opts, Inform) -> action(close_connection, Node, [PidStr, Explanation], _Opts, Inform) ->
Inform("Closing connection ~s", [PidStr]), Inform("Closing connection \"~s\"", [PidStr]),
rpc_call(Node, rabbit_networking, close_connection, rpc_call(Node, rabbit_networking, close_connection,
[rabbit_misc:string_to_pid(PidStr), Explanation]); [rabbit_misc:string_to_pid(PidStr), Explanation]);
action(add_user, Node, Args = [Username, _Password], _Opts, Inform) -> action(add_user, Node, Args = [Username, _Password], _Opts, Inform) ->
Inform("Creating user ~p", [Username]), Inform("Creating user \"~s\"", [Username]),
call(Node, {rabbit_auth_backend_internal, add_user, Args}); call(Node, {rabbit_auth_backend_internal, add_user, Args});
action(delete_user, Node, Args = [_Username], _Opts, Inform) -> action(delete_user, Node, Args = [_Username], _Opts, Inform) ->
Inform("Deleting user ~p", Args), Inform("Deleting user \"~s\"", Args),
call(Node, {rabbit_auth_backend_internal, delete_user, Args}); call(Node, {rabbit_auth_backend_internal, delete_user, Args});
action(change_password, Node, Args = [Username, _Newpassword], _Opts, Inform) -> action(change_password, Node, Args = [Username, _Newpassword], _Opts, Inform) ->
Inform("Changing password for user ~p", [Username]), Inform("Changing password for user \"~s\"", [Username]),
call(Node, {rabbit_auth_backend_internal, change_password, Args}); call(Node, {rabbit_auth_backend_internal, change_password, Args});
action(clear_password, Node, Args = [Username], _Opts, Inform) -> action(clear_password, Node, Args = [Username], _Opts, Inform) ->
Inform("Clearing password for user ~p", [Username]), Inform("Clearing password for user \"~s\"", [Username]),
call(Node, {rabbit_auth_backend_internal, clear_password, Args}); call(Node, {rabbit_auth_backend_internal, clear_password, Args});
action(set_user_tags, Node, [Username | TagsStr], _Opts, Inform) -> action(set_user_tags, Node, [Username | TagsStr], _Opts, Inform) ->
Tags = [list_to_atom(T) || T <- TagsStr], Tags = [list_to_atom(T) || T <- TagsStr],
Inform("Setting tags for user ~p to ~p", [Username, Tags]), Inform("Setting tags for user \"~s\" to ~p", [Username, Tags]),
rpc_call(Node, rabbit_auth_backend_internal, set_tags, rpc_call(Node, rabbit_auth_backend_internal, set_tags,
[list_to_binary(Username), Tags]); [list_to_binary(Username), Tags]);
@ -253,11 +306,11 @@ action(list_users, Node, [], _Opts, Inform) ->
rabbit_auth_backend_internal:user_info_keys()); rabbit_auth_backend_internal:user_info_keys());
action(add_vhost, Node, Args = [_VHostPath], _Opts, Inform) -> action(add_vhost, Node, Args = [_VHostPath], _Opts, Inform) ->
Inform("Creating vhost ~p", Args), Inform("Creating vhost \"~s\"", Args),
call(Node, {rabbit_vhost, add, Args}); call(Node, {rabbit_vhost, add, Args});
action(delete_vhost, Node, Args = [_VHostPath], _Opts, Inform) -> action(delete_vhost, Node, Args = [_VHostPath], _Opts, Inform) ->
Inform("Deleting vhost ~p", Args), Inform("Deleting vhost \"~s\"", Args),
call(Node, {rabbit_vhost, delete, Args}); call(Node, {rabbit_vhost, delete, Args});
action(list_vhosts, Node, Args, _Opts, Inform) -> action(list_vhosts, Node, Args, _Opts, Inform) ->
@ -319,12 +372,12 @@ action(list_consumers, Node, _Args, Opts, Inform) ->
action(trace_on, Node, [], Opts, Inform) -> action(trace_on, Node, [], Opts, Inform) ->
VHost = proplists:get_value(?VHOST_OPT, Opts), VHost = proplists:get_value(?VHOST_OPT, Opts),
Inform("Starting tracing for vhost ~p", [VHost]), Inform("Starting tracing for vhost \"~s\"", [VHost]),
rpc_call(Node, rabbit_trace, start, [list_to_binary(VHost)]); rpc_call(Node, rabbit_trace, start, [list_to_binary(VHost)]);
action(trace_off, Node, [], Opts, Inform) -> action(trace_off, Node, [], Opts, Inform) ->
VHost = proplists:get_value(?VHOST_OPT, Opts), VHost = proplists:get_value(?VHOST_OPT, Opts),
Inform("Stopping tracing for vhost ~p", [VHost]), Inform("Stopping tracing for vhost \"~s\"", [VHost]),
rpc_call(Node, rabbit_trace, stop, [list_to_binary(VHost)]); rpc_call(Node, rabbit_trace, stop, [list_to_binary(VHost)]);
action(set_vm_memory_high_watermark, Node, [Arg], _Opts, Inform) -> action(set_vm_memory_high_watermark, Node, [Arg], _Opts, Inform) ->
@ -337,19 +390,21 @@ action(set_vm_memory_high_watermark, Node, [Arg], _Opts, Inform) ->
action(set_permissions, Node, [Username, CPerm, WPerm, RPerm], Opts, Inform) -> action(set_permissions, Node, [Username, CPerm, WPerm, RPerm], Opts, Inform) ->
VHost = proplists:get_value(?VHOST_OPT, Opts), VHost = proplists:get_value(?VHOST_OPT, Opts),
Inform("Setting permissions for user ~p in vhost ~p", [Username, VHost]), Inform("Setting permissions for user \"~s\" in vhost \"~s\"",
[Username, VHost]),
call(Node, {rabbit_auth_backend_internal, set_permissions, call(Node, {rabbit_auth_backend_internal, set_permissions,
[Username, VHost, CPerm, WPerm, RPerm]}); [Username, VHost, CPerm, WPerm, RPerm]});
action(clear_permissions, Node, [Username], Opts, Inform) -> action(clear_permissions, Node, [Username], Opts, Inform) ->
VHost = proplists:get_value(?VHOST_OPT, Opts), VHost = proplists:get_value(?VHOST_OPT, Opts),
Inform("Clearing permissions for user ~p in vhost ~p", [Username, VHost]), Inform("Clearing permissions for user \"~s\" in vhost \"~s\"",
[Username, VHost]),
call(Node, {rabbit_auth_backend_internal, clear_permissions, call(Node, {rabbit_auth_backend_internal, clear_permissions,
[Username, VHost]}); [Username, VHost]});
action(list_permissions, Node, [], Opts, Inform) -> action(list_permissions, Node, [], Opts, Inform) ->
VHost = proplists:get_value(?VHOST_OPT, Opts), VHost = proplists:get_value(?VHOST_OPT, Opts),
Inform("Listing permissions in vhost ~p", [VHost]), Inform("Listing permissions in vhost \"~s\"", [VHost]),
display_info_list(call(Node, {rabbit_auth_backend_internal, display_info_list(call(Node, {rabbit_auth_backend_internal,
list_vhost_permissions, [VHost]}), list_vhost_permissions, [VHost]}),
rabbit_auth_backend_internal:vhost_perms_info_keys()); rabbit_auth_backend_internal:vhost_perms_info_keys());
@ -405,12 +460,22 @@ wait_for_application(Node, PidFile, Application, Inform) ->
Inform("pid is ~s", [Pid]), Inform("pid is ~s", [Pid]),
wait_for_application(Node, Pid, Application). wait_for_application(Node, Pid, Application).
wait_for_application(Node, Pid, rabbit_and_plugins) ->
wait_for_startup(Node, Pid);
wait_for_application(Node, Pid, Application) -> wait_for_application(Node, Pid, Application) ->
while_process_is_alive(
Node, Pid, fun() -> rabbit_nodes:is_running(Node, Application) end).
wait_for_startup(Node, Pid) ->
while_process_is_alive(
Node, Pid, fun() -> rpc:call(Node, rabbit, await_startup, []) =:= ok end).
while_process_is_alive(Node, Pid, Activity) ->
case process_up(Pid) of case process_up(Pid) of
true -> case rabbit_nodes:is_running(Node, Application) of true -> case Activity() of
true -> ok; true -> ok;
false -> timer:sleep(?EXTERNAL_CHECK_INTERVAL), false -> timer:sleep(?EXTERNAL_CHECK_INTERVAL),
wait_for_application(Node, Pid, Application) while_process_is_alive(Node, Pid, Activity)
end; end;
false -> {error, process_not_running} false -> {error, process_not_running}
end. end.
@ -425,7 +490,7 @@ wait_for_process_death(Pid) ->
read_pid_file(PidFile, Wait) -> read_pid_file(PidFile, Wait) ->
case {file:read_file(PidFile), Wait} of case {file:read_file(PidFile), Wait} of
{{ok, Bin}, _} -> {{ok, Bin}, _} ->
S = string:strip(binary_to_list(Bin), right, $\n), S = re:replace(Bin, "\\s", "", [global, {return, list}]),
try list_to_integer(S) try list_to_integer(S)
catch error:badarg -> catch error:badarg ->
exit({error, {garbage_in_pid_file, PidFile}}) exit({error, {garbage_in_pid_file, PidFile}})

12
src/rabbit_direct.erl
View File

@ -47,16 +47,10 @@
%%---------------------------------------------------------------------------- %%----------------------------------------------------------------------------
boot() -> boot() -> rabbit_sup:start_supervisor_child(
{ok, _} = rabbit_direct_client_sup, rabbit_client_sup,
supervisor2:start_child(
rabbit_sup,
{rabbit_direct_client_sup,
{rabbit_client_sup, start_link,
[{local, rabbit_direct_client_sup}, [{local, rabbit_direct_client_sup},
{rabbit_channel_sup, start_link, []}]}, {rabbit_channel_sup, start_link, []}]).
transient, infinity, supervisor, [rabbit_client_sup]}),
ok.
force_event_refresh() -> force_event_refresh() ->
[Pid ! force_event_refresh || Pid<- list()], [Pid ! force_event_refresh || Pid<- list()],

5
src/rabbit_disk_monitor.erl
View File

@ -178,8 +178,9 @@ parse_free_unix(CommandResult) ->
parse_free_win32(CommandResult) -> parse_free_win32(CommandResult) ->
LastLine = lists:last(string:tokens(CommandResult, "\r\n")), LastLine = lists:last(string:tokens(CommandResult, "\r\n")),
[_, _Dir, Free, "bytes", "free"] = string:tokens(LastLine, " "), {match, [Free]} = re:run(lists:reverse(LastLine), "(\\d+)",
list_to_integer(Free). [{capture, all_but_first, list}]),
list_to_integer(lists:reverse(Free)).
interpret_limit({mem_relative, R}) -> interpret_limit({mem_relative, R}) ->
round(R * vm_memory_monitor:get_total_memory()); round(R * vm_memory_monitor:get_total_memory());

17
src/rabbit_mirror_queue_slave_sup.erl
View File

@ -18,7 +18,7 @@
-behaviour(supervisor2). -behaviour(supervisor2).
-export([start/0, start_link/0, start_child/2]). -export([start_link/0, start_child/2]).
-export([init/1]). -export([init/1]).
@ -26,20 +26,9 @@
-define(SERVER, ?MODULE). -define(SERVER, ?MODULE).
start() -> start_link() -> supervisor2:start_link({local, ?SERVER}, ?MODULE, []).
{ok, _} =
supervisor2:start_child(
rabbit_sup,
{rabbit_mirror_queue_slave_sup,
{rabbit_mirror_queue_slave_sup, start_link, []},
transient, infinity, supervisor, [rabbit_mirror_queue_slave_sup]}),
ok.
start_link() -> start_child(Node, Args) -> supervisor2:start_child({?SERVER, Node}, Args).
supervisor2:start_link({local, ?SERVER}, ?MODULE, []).
start_child(Node, Args) ->
supervisor2:start_child({?SERVER, Node}, Args).
init([]) -> init([]) ->
{ok, {{simple_one_for_one_terminate, 10, 10}, {ok, {{simple_one_for_one_terminate, 10, 10},

166
src/rabbit_misc.erl
View File

@ -19,7 +19,7 @@
-include("rabbit_framing.hrl"). -include("rabbit_framing.hrl").
-export([method_record_type/1, polite_pause/0, polite_pause/1]). -export([method_record_type/1, polite_pause/0, polite_pause/1]).
-export([die/1, frame_error/2, amqp_error/4, -export([die/1, frame_error/2, amqp_error/4, quit/1, quit/2,
protocol_error/3, protocol_error/4, protocol_error/1]). protocol_error/3, protocol_error/4, protocol_error/1]).
-export([not_found/1, assert_args_equivalence/4]). -export([not_found/1, assert_args_equivalence/4]).
-export([dirty_read/1]). -export([dirty_read/1]).
@ -42,14 +42,13 @@
-export([dirty_read_all/1, dirty_foreach_key/2, dirty_dump_log/1]). -export([dirty_read_all/1, dirty_foreach_key/2, dirty_dump_log/1]).
-export([format/2, format_many/1, format_stderr/2]). -export([format/2, format_many/1, format_stderr/2]).
-export([with_local_io/1, local_info_msg/2]). -export([with_local_io/1, local_info_msg/2]).
-export([start_applications/1, stop_applications/1]).
-export([unfold/2, ceil/1, queue_fold/3]). -export([unfold/2, ceil/1, queue_fold/3]).
-export([sort_field_table/1]). -export([sort_field_table/1]).
-export([pid_to_string/1, string_to_pid/1]). -export([pid_to_string/1, string_to_pid/1]).
-export([version_compare/2, version_compare/3]). -export([version_compare/2, version_compare/3]).
-export([dict_cons/3, orddict_cons/3, gb_trees_cons/3]). -export([dict_cons/3, orddict_cons/3, gb_trees_cons/3]).
-export([gb_trees_fold/3, gb_trees_foreach/2]). -export([gb_trees_fold/3, gb_trees_foreach/2]).
-export([get_options/2]). -export([parse_arguments/3]).
-export([all_module_attributes/1, build_acyclic_graph/3]). -export([all_module_attributes/1, build_acyclic_graph/3]).
-export([now_ms/0]). -export([now_ms/0]).
-export([const_ok/0, const/1]). -export([const_ok/0, const/1]).
@ -59,7 +58,6 @@
-export([format_message_queue/2]). -export([format_message_queue/2]).
-export([append_rpc_all_nodes/4]). -export([append_rpc_all_nodes/4]).
-export([multi_call/2]). -export([multi_call/2]).
-export([quit/1]).
-export([os_cmd/1]). -export([os_cmd/1]).
-export([gb_sets_difference/2]). -export([gb_sets_difference/2]).
@ -72,7 +70,7 @@
-type(ok_or_error() :: rabbit_types:ok_or_error(any())). -type(ok_or_error() :: rabbit_types:ok_or_error(any())).
-type(thunk(T) :: fun(() -> T)). -type(thunk(T) :: fun(() -> T)).
-type(resource_name() :: binary()). -type(resource_name() :: binary()).
-type(optdef() :: {flag, string()} | {option, string(), any()}). -type(optdef() :: flag | {option, string()}).
-type(channel_or_connection_exit() -type(channel_or_connection_exit()
:: rabbit_types:channel_exit() | rabbit_types:connection_exit()). :: rabbit_types:channel_exit() | rabbit_types:connection_exit()).
-type(digraph_label() :: term()). -type(digraph_label() :: term()).
@ -87,6 +85,10 @@
-spec(polite_pause/1 :: (non_neg_integer()) -> 'done'). -spec(polite_pause/1 :: (non_neg_integer()) -> 'done').
-spec(die/1 :: -spec(die/1 ::
(rabbit_framing:amqp_exception()) -> channel_or_connection_exit()). (rabbit_framing:amqp_exception()) -> channel_or_connection_exit()).
-spec(quit/1 :: (integer()) -> no_return()).
-spec(quit/2 :: (string(), [term()]) -> no_return()).
-spec(frame_error/2 :: (rabbit_framing:amqp_method_name(), binary()) -spec(frame_error/2 :: (rabbit_framing:amqp_method_name(), binary())
-> rabbit_types:connection_exit()). -> rabbit_types:connection_exit()).
-spec(amqp_error/4 :: -spec(amqp_error/4 ::
@ -163,8 +165,6 @@
-spec(format_stderr/2 :: (string(), [any()]) -> 'ok'). -spec(format_stderr/2 :: (string(), [any()]) -> 'ok').
-spec(with_local_io/1 :: (fun (() -> A)) -> A). -spec(with_local_io/1 :: (fun (() -> A)) -> A).
-spec(local_info_msg/2 :: (string(), [any()]) -> 'ok'). -spec(local_info_msg/2 :: (string(), [any()]) -> 'ok').
-spec(start_applications/1 :: ([atom()]) -> 'ok').
-spec(stop_applications/1 :: ([atom()]) -> 'ok').
-spec(unfold/2 :: (fun ((A) -> ({'true', B, A} | 'false')), A) -> {[B], A}). -spec(unfold/2 :: (fun ((A) -> ({'true', B, A} | 'false')), A) -> {[B], A}).
-spec(ceil/1 :: (number()) -> integer()). -spec(ceil/1 :: (number()) -> integer()).
-spec(queue_fold/3 :: (fun ((any(), B) -> B), B, queue()) -> B). -spec(queue_fold/3 :: (fun ((any(), B) -> B), B, queue()) -> B).
@ -182,8 +182,12 @@
-spec(gb_trees_fold/3 :: (fun ((any(), any(), A) -> A), A, gb_tree()) -> A). -spec(gb_trees_fold/3 :: (fun ((any(), any(), A) -> A), A, gb_tree()) -> A).
-spec(gb_trees_foreach/2 :: -spec(gb_trees_foreach/2 ::
(fun ((any(), any()) -> any()), gb_tree()) -> 'ok'). (fun ((any(), any()) -> any()), gb_tree()) -> 'ok').
-spec(get_options/2 :: ([optdef()], [string()]) -spec(parse_arguments/3 ::
-> {[string()], [{string(), any()}]}). ([{atom(), [{string(), optdef()}]} | atom()],
[{string(), optdef()}],
[string()])
-> {'ok', {atom(), [{string(), string()}], [string()]}} |
'no_command').
-spec(all_module_attributes/1 :: (atom()) -> [{atom(), [term()]}]). -spec(all_module_attributes/1 :: (atom()) -> [{atom(), [term()]}]).
-spec(build_acyclic_graph/3 :: -spec(build_acyclic_graph/3 ::
(graph_vertex_fun(), graph_edge_fun(), [{atom(), [term()]}]) (graph_vertex_fun(), graph_edge_fun(), [{atom(), [term()]}])
@ -206,7 +210,6 @@
-spec(append_rpc_all_nodes/4 :: ([node()], atom(), atom(), [any()]) -> [any()]). -spec(append_rpc_all_nodes/4 :: ([node()], atom(), atom(), [any()]) -> [any()]).
-spec(multi_call/2 :: -spec(multi_call/2 ::
([pid()], any()) -> {[{pid(), any()}], [{pid(), any()}]}). ([pid()], any()) -> {[{pid(), any()}], [{pid(), any()}]}).
-spec(quit/1 :: (integer() | string()) -> no_return()).
-spec(os_cmd/1 :: (string()) -> string()). -spec(os_cmd/1 :: (string()) -> string()).
-spec(gb_sets_difference/2 :: (gb_set(), gb_set()) -> gb_set()). -spec(gb_sets_difference/2 :: (gb_set(), gb_set()) -> gb_set()).
@ -387,6 +390,28 @@ report_coverage_percentage(File, Cov, NotCov, Mod) ->
confirm_to_sender(Pid, MsgSeqNos) -> confirm_to_sender(Pid, MsgSeqNos) ->
gen_server2:cast(Pid, {confirm, MsgSeqNos, self()}). gen_server2:cast(Pid, {confirm, MsgSeqNos, self()}).
%%
%% @doc Halts the emulator after printing out an error message io-formatted with
%% the supplied arguments. The exit status of the beam process will be set to 1.
%%
quit(Fmt, Args) ->
io:format("ERROR: " ++ Fmt ++ "~n", Args),
quit(1).
%%
%% @doc Halts the emulator returning the given status code to the os.
%% On Windows this function will block indefinitely so as to give the io
%% subsystem time to flush stdout completely.
%%
quit(Status) ->
case os:type() of
{unix, _} -> halt(Status);
{win32, _} -> init:stop(Status),
receive
after infinity -> ok
end
end.
throw_on_error(E, Thunk) -> throw_on_error(E, Thunk) ->
case Thunk() of case Thunk() of
{error, Reason} -> throw({E, Reason}); {error, Reason} -> throw({E, Reason});
@ -589,34 +614,6 @@ with_local_io(Fun) ->
local_info_msg(Format, Args) -> local_info_msg(Format, Args) ->
with_local_io(fun () -> error_logger:info_msg(Format, Args) end). with_local_io(fun () -> error_logger:info_msg(Format, Args) end).
manage_applications(Iterate, Do, Undo, SkipError, ErrorTag, Apps) ->
Iterate(fun (App, Acc) ->
case Do(App) of
ok -> [App | Acc];
{error, {SkipError, _}} -> Acc;
{error, Reason} ->
lists:foreach(Undo, Acc),
throw({error, {ErrorTag, App, Reason}})
end
end, [], Apps),
ok.
start_applications(Apps) ->
manage_applications(fun lists:foldl/3,
fun application:start/1,
fun application:stop/1,
already_started,
cannot_start_application,
Apps).
stop_applications(Apps) ->
manage_applications(fun lists:foldr/3,
fun application:stop/1,
fun application:start/1,
not_started,
cannot_stop_application,
Apps).
unfold(Fun, Init) -> unfold(Fun, Init) ->
unfold(Fun, [], Init). unfold(Fun, [], Init).
@ -736,39 +733,63 @@ gb_trees_fold1(Fun, Acc, {Key, Val, It}) ->
gb_trees_foreach(Fun, Tree) -> gb_trees_foreach(Fun, Tree) ->
gb_trees_fold(fun (Key, Val, Acc) -> Fun(Key, Val), Acc end, ok, Tree). gb_trees_fold(fun (Key, Val, Acc) -> Fun(Key, Val), Acc end, ok, Tree).
%% Separate flags and options from arguments. %% Takes:
%% get_options([{flag, "-q"}, {option, "-p", "/"}], %% * A list of [{atom(), [{string(), optdef()]} | atom()], where the atom()s
%% ["set_permissions","-p","/","guest", %% are the accepted commands and the optional [string()] is the list of
%% "-q",".*",".*",".*"]) %% accepted options for that command
%% == {["set_permissions","guest",".*",".*",".*"], %% * A list [{string(), optdef()}] of options valid for all commands
%% [{"-q",true},{"-p","/"}]} %% * The list of arguments given by the user
get_options(Defs, As) -> %%
lists:foldl(fun(Def, {AsIn, RsIn}) -> %% Returns either {ok, {atom(), [{string(), string()}], [string()]} which are
{K, {AsOut, V}} = %% respectively the command, the key-value pairs of the options and the leftover
case Def of %% arguments; or no_command if no command could be parsed.
{flag, Key} -> parse_arguments(Commands, GlobalDefs, As) ->
{Key, get_flag(Key, AsIn)}; lists:foldl(maybe_process_opts(GlobalDefs, As), no_command, Commands).
{option, Key, Default} ->
{Key, get_option(Key, Default, AsIn)}
end,
{AsOut, [{K, V} | RsIn]}
end, {As, []}, Defs).
get_option(K, _Default, [K, V | As]) -> maybe_process_opts(GDefs, As) ->
{As, V}; fun({C, Os}, no_command) ->
get_option(K, Default, [Nk | As]) -> process_opts(atom_to_list(C), dict:from_list(GDefs ++ Os), As);
{As1, V} = get_option(K, Default, As), (C, no_command) ->
{[Nk | As1], V}; (maybe_process_opts(GDefs, As))({C, []}, no_command);
get_option(_, Default, As) -> (_, {ok, Res}) ->
{As, Default}. {ok, Res}
end.
get_flag(K, [K | As]) -> process_opts(C, Defs, As0) ->
{As, true}; KVs0 = dict:map(fun (_, flag) -> false;
get_flag(K, [Nk | As]) -> (_, {option, V}) -> V
{As1, V} = get_flag(K, As), end, Defs),
{[Nk | As1], V}; process_opts(Defs, C, As0, not_found, KVs0, []).
get_flag(_, []) ->
{[], false}. %% Consume flags/options until you find the correct command. If there are no
%% arguments or the first argument is not the command we're expecting, fail.
%% Arguments to this are: definitions, cmd we're looking for, args we
%% haven't parsed, whether we have found the cmd, options we've found,
%% plain args we've found.
process_opts(_Defs, C, [], found, KVs, Outs) ->
{ok, {list_to_atom(C), dict:to_list(KVs), lists:reverse(Outs)}};
process_opts(_Defs, _C, [], not_found, _, _) ->
no_command;
process_opts(Defs, C, [A | As], Found, KVs, Outs) ->
OptType = case dict:find(A, Defs) of
error -> none;
{ok, flag} -> flag;
{ok, {option, _}} -> option
end,
case {OptType, C, Found} of
{flag, _, _} -> process_opts(
Defs, C, As, Found, dict:store(A, true, KVs),
Outs);
{option, _, _} -> case As of
[] -> no_command;
[V | As1] -> process_opts(
Defs, C, As1, Found,
dict:store(A, V, KVs), Outs)
end;
{none, A, _} -> process_opts(Defs, C, As, found, KVs, Outs);
{none, _, found} -> process_opts(Defs, C, As, found, KVs, [A | Outs]);
{none, _, _} -> no_command
end.
now_ms() -> now_ms() ->
timer:now_diff(now(), {0,0,0}) div 1000. timer:now_diff(now(), {0,0,0}) div 1000.
@ -909,13 +930,6 @@ receive_multi_call([{Mref, Pid} | MonitorPids], Good, Bad) ->
receive_multi_call(MonitorPids, Good, [{Pid, Reason} | Bad]) receive_multi_call(MonitorPids, Good, [{Pid, Reason} | Bad])
end. end.
%% the slower shutdown on windows required to flush stdout
quit(Status) ->
case os:type() of
{unix, _} -> halt(Status);
{win32, _} -> init:stop(Status)
end.
os_cmd(Command) -> os_cmd(Command) ->
Exec = hd(string:tokens(Command, " ")), Exec = hd(string:tokens(Command, " ")),
case os:find_executable(Exec) of case os:find_executable(Exec) of

10
src/rabbit_net.erl
View File

@ -20,7 +20,7 @@
-export([is_ssl/1, ssl_info/1, controlling_process/2, getstat/2, -export([is_ssl/1, ssl_info/1, controlling_process/2, getstat/2,
recv/1, async_recv/3, port_command/2, getopts/2, setopts/2, send/2, recv/1, async_recv/3, port_command/2, getopts/2, setopts/2, send/2,
close/1, maybe_fast_close/1, sockname/1, peername/1, peercert/1, close/1, maybe_fast_close/1, sockname/1, peername/1, peercert/1,
connection_string/2]). tune_buffer_size/1, connection_string/2]).
%%--------------------------------------------------------------------------- %%---------------------------------------------------------------------------
@ -69,6 +69,7 @@
-spec(peercert/1 :: -spec(peercert/1 ::
(socket()) (socket())
-> 'nossl' | ok_val_or_error(rabbit_ssl:certificate())). -> 'nossl' | ok_val_or_error(rabbit_ssl:certificate())).
-spec(tune_buffer_size/1 :: (socket()) -> ok_or_any_error()).
-spec(connection_string/2 :: -spec(connection_string/2 ::
(socket(), 'inbound' | 'outbound') -> ok_val_or_error(string())). (socket(), 'inbound' | 'outbound') -> ok_val_or_error(string())).
@ -159,6 +160,13 @@ peername(Sock) when is_port(Sock) -> inet:peername(Sock).
peercert(Sock) when ?IS_SSL(Sock) -> ssl:peercert(Sock#ssl_socket.ssl); peercert(Sock) when ?IS_SSL(Sock) -> ssl:peercert(Sock#ssl_socket.ssl);
peercert(Sock) when is_port(Sock) -> nossl. peercert(Sock) when is_port(Sock) -> nossl.
tune_buffer_size(Sock) ->
case getopts(Sock, [sndbuf, recbuf, buffer]) of
{ok, BufSizes} -> BufSz = lists:max([Sz || {_Opt, Sz} <- BufSizes]),
setopts(Sock, [{buffer, BufSz}]);
Err -> Err
end.
connection_string(Sock, Direction) -> connection_string(Sock, Direction) ->
{From, To} = case Direction of {From, To} = case Direction of
inbound -> {fun peername/1, fun sockname/1}; inbound -> {fun peername/1, fun sockname/1};

15
src/rabbit_networking.erl
View File

@ -136,18 +136,13 @@ boot_ssl() ->
ok ok
end. end.
start() -> start() -> rabbit_sup:start_supervisor_child(
{ok,_} = supervisor2:start_child( rabbit_tcp_client_sup, rabbit_client_sup,
rabbit_sup, [{local, rabbit_tcp_client_sup},
{rabbit_tcp_client_sup, {rabbit_connection_sup,start_link,[]}]).
{rabbit_client_sup, start_link,
[{local, rabbit_tcp_client_sup},
{rabbit_connection_sup,start_link,[]}]},
transient, infinity, supervisor, [rabbit_client_sup]}),
ok.
ensure_ssl() -> ensure_ssl() ->
ok = rabbit_misc:start_applications([crypto, public_key, ssl]), ok = app_utils:start_applications([crypto, public_key, ssl]),
{ok, SslOptsConfig} = application:get_env(rabbit, ssl_options), {ok, SslOptsConfig} = application:get_env(rabbit, ssl_options),
% unknown_ca errors are silently ignored prior to R14B unless we % unknown_ca errors are silently ignored prior to R14B unless we

395
src/rabbit_plugins.erl
View File

@ -17,146 +17,57 @@
-module(rabbit_plugins). -module(rabbit_plugins).
-include("rabbit.hrl"). -include("rabbit.hrl").
-export([start/0, stop/0, find_plugins/1, read_enabled_plugins/1, -export([setup/0, active/0, read_enabled/1,
lookup_plugins/2, calculate_required_plugins/2, plugin_names/1]). list/1, dependencies/3]).
-define(VERBOSE_OPT, "-v"). -define(VERBOSE_DEF, {?VERBOSE_OPT, flag}).
-define(MINIMAL_OPT, "-m"). -define(MINIMAL_DEF, {?MINIMAL_OPT, flag}).
-define(ENABLED_OPT, "-E"). -define(ENABLED_DEF, {?ENABLED_OPT, flag}).
-define(ENABLED_ALL_OPT, "-e"). -define(ENABLED_ALL_DEF, {?ENABLED_ALL_OPT, flag}).
-define(GLOBAL_DEFS, []).
-define(COMMANDS,
[{list, [?VERBOSE_DEF, ?MINIMAL_DEF, ?ENABLED_DEF, ?ENABLED_ALL_DEF]},
enable,
disable]).
%%---------------------------------------------------------------------------- %%----------------------------------------------------------------------------
-ifdef(use_specs). -ifdef(use_specs).
-spec(start/0 :: () -> no_return()). -spec(setup/0 :: () -> [atom()]).
-spec(stop/0 :: () -> 'ok'). -spec(active/0 :: () -> [atom()]).
-spec(find_plugins/1 :: (file:filename()) -> [#plugin{}]). -spec(list/1 :: (string()) -> [#plugin{}]).
-spec(read_enabled_plugins/1 :: (file:filename()) -> [atom()]). -spec(read_enabled/1 :: (file:filename()) -> [atom()]).
-spec(lookup_plugins/2 :: ([atom()], [#plugin{}]) -> [#plugin{}]). -spec(dependencies/3 ::
-spec(calculate_required_plugins/2 :: ([atom()], [#plugin{}]) -> [atom()]). (boolean(), [atom()], [#plugin{}]) -> [atom()]).
-spec(plugin_names/1 :: ([#plugin{}]) -> [atom()]).
-endif. -endif.
%%---------------------------------------------------------------------------- %%----------------------------------------------------------------------------
start() -> %%
{ok, [[PluginsFile|_]|_]} = %% @doc Prepares the file system and installs all enabled plugins.
init:get_argument(enabled_plugins_file), %%
{ok, [[PluginsDir|_]|_]} = init:get_argument(plugins_dist_dir), setup() ->
{[Command0 | Args], Opts} = {ok, PluginDir} = application:get_env(rabbit, plugins_dir),
case rabbit_misc:get_options([{flag, ?VERBOSE_OPT}, {ok, ExpandDir} = application:get_env(rabbit, plugins_expand_dir),
{flag, ?MINIMAL_OPT}, {ok, EnabledPluginsFile} = application:get_env(rabbit,
{flag, ?ENABLED_OPT}, enabled_plugins_file),
{flag, ?ENABLED_ALL_OPT}], prepare_plugins(EnabledPluginsFile, PluginDir, ExpandDir),
init:get_plain_arguments()) of [prepare_dir_plugin(PluginName) ||
{[], _Opts} -> usage(); PluginName <- filelib:wildcard(ExpandDir ++ "/*/ebin/*.app")].
CmdArgsAndOpts -> CmdArgsAndOpts
end,
Command = list_to_atom(Command0),
PrintInvalidCommandError =
fun () ->
print_error("invalid command '~s'",
[string:join([atom_to_list(Command) | Args], " ")])
end,
case catch action(Command, Args, Opts, PluginsFile, PluginsDir) of %% @doc Lists the plugins which are currently running.
ok -> active() ->
rabbit_misc:quit(0); {ok, ExpandDir} = application:get_env(rabbit, plugins_expand_dir),
{'EXIT', {function_clause, [{?MODULE, action, _} | _]}} -> InstalledPlugins = [ P#plugin.name || P <- list(ExpandDir) ],
PrintInvalidCommandError(), [App || {App, _, _} <- application:which_applications(),
usage(); lists:member(App, InstalledPlugins)].
{'EXIT', {function_clause, [{?MODULE, action, _, _} | _]}} ->
PrintInvalidCommandError(),
usage();
{error, Reason} ->
print_error("~p", [Reason]),
rabbit_misc:quit(2);
{error_string, Reason} ->
print_error("~s", [Reason]),
rabbit_misc:quit(2);
Other ->
print_error("~p", [Other]),
rabbit_misc:quit(2)
end.
stop() -> %% @doc Get the list of plugins which are ready to be enabled.
ok. list(PluginsDir) ->
print_error(Format, Args) ->
rabbit_misc:format_stderr("Error: " ++ Format ++ "~n", Args).
usage() ->
io:format("~s", [rabbit_plugins_usage:usage()]),
rabbit_misc:quit(1).
%%----------------------------------------------------------------------------
action(list, [], Opts, PluginsFile, PluginsDir) ->
action(list, [".*"], Opts, PluginsFile, PluginsDir);
action(list, [Pat], Opts, PluginsFile, PluginsDir) ->
format_plugins(Pat, Opts, PluginsFile, PluginsDir);
action(enable, ToEnable0, _Opts, PluginsFile, PluginsDir) ->
case ToEnable0 of
[] -> throw({error_string, "Not enough arguments for 'enable'"});
_ -> ok
end,
AllPlugins = find_plugins(PluginsDir),
Enabled = read_enabled_plugins(PluginsFile),
ImplicitlyEnabled = calculate_required_plugins(Enabled, AllPlugins),
ToEnable = [list_to_atom(Name) || Name <- ToEnable0],
Missing = ToEnable -- plugin_names(AllPlugins),
case Missing of
[] -> ok;
_ -> throw({error_string,
fmt_list("The following plugins could not be found:",
Missing)})
end,
NewEnabled = lists:usort(Enabled ++ ToEnable),
write_enabled_plugins(PluginsFile, NewEnabled),
NewImplicitlyEnabled = calculate_required_plugins(NewEnabled, AllPlugins),
maybe_warn_mochiweb(NewImplicitlyEnabled),
case NewEnabled -- ImplicitlyEnabled of
[] -> io:format("Plugin configuration unchanged.~n");
_ -> print_list("The following plugins have been enabled:",
NewImplicitlyEnabled -- ImplicitlyEnabled),
report_change()
end;
action(disable, ToDisable0, _Opts, PluginsFile, PluginsDir) ->
case ToDisable0 of
[] -> throw({error_string, "Not enough arguments for 'disable'"});
_ -> ok
end,
ToDisable = [list_to_atom(Name) || Name <- ToDisable0],
Enabled = read_enabled_plugins(PluginsFile),
AllPlugins = find_plugins(PluginsDir),
Missing = ToDisable -- plugin_names(AllPlugins),
case Missing of
[] -> ok;
_ -> print_list("Warning: the following plugins could not be found:",
Missing)
end,
ToDisableDeps = calculate_dependencies(true, ToDisable, AllPlugins),
NewEnabled = Enabled -- ToDisableDeps,
case length(Enabled) =:= length(NewEnabled) of
true -> io:format("Plugin configuration unchanged.~n");
false -> ImplicitlyEnabled =
calculate_required_plugins(Enabled, AllPlugins),
NewImplicitlyEnabled =
calculate_required_plugins(NewEnabled, AllPlugins),
print_list("The following plugins have been disabled:",
ImplicitlyEnabled -- NewImplicitlyEnabled),
write_enabled_plugins(PluginsFile, NewEnabled),
report_change()
end.
%%----------------------------------------------------------------------------
%% Get the #plugin{}s ready to be enabled.
find_plugins(PluginsDir) ->
EZs = [{ez, EZ} || EZ <- filelib:wildcard("*.ez", PluginsDir)], EZs = [{ez, EZ} || EZ <- filelib:wildcard("*.ez", PluginsDir)],
FreeApps = [{app, App} || FreeApps = [{app, App} ||
App <- filelib:wildcard("*/ebin/*.app", PluginsDir)], App <- filelib:wildcard("*/ebin/*.app", PluginsDir)],
@ -175,6 +86,91 @@ find_plugins(PluginsDir) ->
end, end,
Plugins. Plugins.
%% @doc Read the list of enabled plugins from the supplied term file.
read_enabled(PluginsFile) ->
case rabbit_file:read_term_file(PluginsFile) of
{ok, [Plugins]} -> Plugins;
{ok, []} -> [];
{ok, [_|_]} -> throw({error, {malformed_enabled_plugins_file,
PluginsFile}});
{error, enoent} -> [];
{error, Reason} -> throw({error, {cannot_read_enabled_plugins_file,
PluginsFile, Reason}})
end.
%%
%% @doc Calculate the dependency graph from <i>Sources</i>.
%% When Reverse =:= true the bottom/leaf level applications are returned in
%% the resulting list, otherwise they're skipped.
%%
dependencies(Reverse, Sources, AllPlugins) ->
{ok, G} = rabbit_misc:build_acyclic_graph(
fun (App, _Deps) -> [{App, App}] end,
fun (App, Deps) -> [{App, Dep} || Dep <- Deps] end,
[{Name, Deps}
|| #plugin{name = Name, dependencies = Deps} <- AllPlugins]),
Dests = case Reverse of
false -> digraph_utils:reachable(Sources, G);
true -> digraph_utils:reaching(Sources, G)
end,
true = digraph:delete(G),
Dests.
%%----------------------------------------------------------------------------
prepare_plugins(EnabledPluginsFile, PluginsDistDir, DestDir) ->
AllPlugins = list(PluginsDistDir),
Enabled = read_enabled(EnabledPluginsFile),
ToUnpack = dependencies(false, Enabled, AllPlugins),
ToUnpackPlugins = lookup_plugins(ToUnpack, AllPlugins),
Missing = Enabled -- plugin_names(ToUnpackPlugins),
case Missing of
[] -> ok;
_ -> io:format("Warning: the following enabled plugins were "
"not found: ~p~n", [Missing])
end,
%% Eliminate the contents of the destination directory
case delete_recursively(DestDir) of
ok -> ok;
{error, E} -> rabbit_misc:quit("Could not delete dir ~s (~p)",
[DestDir, E])
end,
case filelib:ensure_dir(DestDir ++ "/") of
ok -> ok;
{error, E2} -> rabbit_misc:quit("Could not create dir ~s (~p)",
[DestDir, E2])
end,
[prepare_plugin(Plugin, DestDir) || Plugin <- ToUnpackPlugins].
prepare_dir_plugin(PluginAppDescFn) ->
%% Add the plugin ebin directory to the load path
PluginEBinDirN = filename:dirname(PluginAppDescFn),
code:add_path(PluginEBinDirN),
%% We want the second-last token
NameTokens = string:tokens(PluginAppDescFn,"/."),
PluginNameString = lists:nth(length(NameTokens) - 1, NameTokens),
list_to_atom(PluginNameString).
%%----------------------------------------------------------------------------
delete_recursively(Fn) ->
case rabbit_file:recursive_delete([Fn]) of
ok -> ok;
{error, {Path, E}} -> {error, {cannot_delete, Path, E}};
Error -> Error
end.
prepare_plugin(#plugin{type = ez, location = Location}, PluginDestDir) ->
zip:unzip(Location, [{cwd, PluginDestDir}]);
prepare_plugin(#plugin{type = dir, name = Name, location = Location},
PluginsDestDir) ->
rabbit_file:recursive_copy(Location,
filename:join([PluginsDestDir, Name])).
%% Get the #plugin{} from an .ez. %% Get the #plugin{} from an .ez.
get_plugin_info(Base, {ez, EZ0}) -> get_plugin_info(Base, {ez, EZ0}) ->
EZ = filename:join([Base, EZ0]), EZ = filename:join([Base, EZ0]),
@ -234,82 +230,6 @@ parse_binary(Bin) ->
Err -> {error, {invalid_app, Err}} Err -> {error, {invalid_app, Err}}
end. end.
%% Pretty print a list of plugins.
format_plugins(Pattern, Opts, PluginsFile, PluginsDir) ->
Verbose = proplists:get_bool(?VERBOSE_OPT, Opts),
Minimal = proplists:get_bool(?MINIMAL_OPT, Opts),
Format = case {Verbose, Minimal} of
{false, false} -> normal;
{true, false} -> verbose;
{false, true} -> minimal;
{true, true} -> throw({error_string,
"Cannot specify -m and -v together"})
end,
OnlyEnabled = proplists:get_bool(?ENABLED_OPT, Opts),
OnlyEnabledAll = proplists:get_bool(?ENABLED_ALL_OPT, Opts),
AvailablePlugins = find_plugins(PluginsDir),
EnabledExplicitly = read_enabled_plugins(PluginsFile),
EnabledImplicitly =
calculate_required_plugins(EnabledExplicitly, AvailablePlugins) --
EnabledExplicitly,
{ok, RE} = re:compile(Pattern),
Plugins = [ Plugin ||
Plugin = #plugin{name = Name} <- AvailablePlugins,
re:run(atom_to_list(Name), RE, [{capture, none}]) =:= match,
if OnlyEnabled -> lists:member(Name, EnabledExplicitly);
true -> true
end,
if OnlyEnabledAll ->
lists:member(Name, EnabledImplicitly) or
lists:member(Name, EnabledExplicitly);
true ->
true
end],
Plugins1 = usort_plugins(Plugins),
MaxWidth = lists:max([length(atom_to_list(Name)) ||
#plugin{name = Name} <- Plugins1] ++ [0]),
[format_plugin(P, EnabledExplicitly, EnabledImplicitly, Format,
MaxWidth) || P <- Plugins1],
ok.
format_plugin(#plugin{name = Name, version = Version,
description = Description, dependencies = Deps},
EnabledExplicitly, EnabledImplicitly, Format, MaxWidth) ->
Glyph = case {lists:member(Name, EnabledExplicitly),
lists:member(Name, EnabledImplicitly)} of
{true, false} -> "[E]";
{false, true} -> "[e]";
_ -> "[ ]"
end,
case Format of
minimal -> io:format("~s~n", [Name]);
normal -> io:format("~s ~-" ++ integer_to_list(MaxWidth) ++
"w ~s~n", [Glyph, Name, Version]);
verbose -> io:format("~s ~w~n", [Glyph, Name]),
io:format(" Version: \t~s~n", [Version]),
case Deps of
[] -> ok;
_ -> io:format(" Dependencies:\t~p~n", [Deps])
end,
io:format(" Description:\t~s~n", [Description]),
io:format("~n")
end.
print_list(Header, Plugins) ->
io:format(fmt_list(Header, Plugins)).
fmt_list(Header, Plugins) ->
lists:flatten(
[Header, $\n, [io_lib:format(" ~s~n", [P]) || P <- Plugins]]).
usort_plugins(Plugins) ->
lists:usort(fun plugins_cmp/2, Plugins).
plugins_cmp(#plugin{name = N1, version = V1},
#plugin{name = N2, version = V2}) ->
{N1, V1} =< {N2, V2}.
%% Filter out applications that can be loaded *right now*. %% Filter out applications that can be loaded *right now*.
filter_applications(Applications) -> filter_applications(Applications) ->
[Application || Application <- Applications, [Application || Application <- Applications,
@ -332,72 +252,3 @@ plugin_names(Plugins) ->
%% Find plugins by name in a list of plugins. %% Find plugins by name in a list of plugins.
lookup_plugins(Names, AllPlugins) -> lookup_plugins(Names, AllPlugins) ->
[P || P = #plugin{name = Name} <- AllPlugins, lists:member(Name, Names)]. [P || P = #plugin{name = Name} <- AllPlugins, lists:member(Name, Names)].
%% Read the enabled plugin names from disk.
read_enabled_plugins(PluginsFile) ->
case rabbit_file:read_term_file(PluginsFile) of
{ok, [Plugins]} -> Plugins;
{ok, []} -> [];
{ok, [_|_]} -> throw({error, {malformed_enabled_plugins_file,
PluginsFile}});
{error, enoent} -> [];
{error, Reason} -> throw({error, {cannot_read_enabled_plugins_file,
PluginsFile, Reason}})
end.
%% Write the enabled plugin names on disk.
write_enabled_plugins(PluginsFile, Plugins) ->
case rabbit_file:write_term_file(PluginsFile, [Plugins]) of
ok -> ok;
{error, Reason} -> throw({error, {cannot_write_enabled_plugins_file,
PluginsFile, Reason}})
end.
calculate_required_plugins(Sources, AllPlugins) ->
calculate_dependencies(false, Sources, AllPlugins).
calculate_dependencies(Reverse, Sources, AllPlugins) ->
{ok, G} = rabbit_misc:build_acyclic_graph(
fun (App, _Deps) -> [{App, App}] end,
fun (App, Deps) -> [{App, Dep} || Dep <- Deps] end,
[{Name, Deps}
|| #plugin{name = Name, dependencies = Deps} <- AllPlugins]),
Dests = case Reverse of
false -> digraph_utils:reachable(Sources, G);
true -> digraph_utils:reaching(Sources, G)
end,
true = digraph:delete(G),
Dests.
maybe_warn_mochiweb(Enabled) ->
V = erlang:system_info(otp_release),
case lists:member(mochiweb, Enabled) andalso V < "R13B01" of
true ->
Stars = string:copies("*", 80),
io:format("~n~n~s~n"
" Warning: Mochiweb enabled and Erlang version ~s "
"detected.~n"
" Enabling plugins that depend on Mochiweb is not "
"supported on this Erlang~n"
" version. At least R13B01 is required.~n~n"
" RabbitMQ will not start successfully in this "
"configuration. You *must*~n"
" disable the Mochiweb plugin, or upgrade Erlang.~n"
"~s~n~n~n", [Stars, V, Stars]);
false ->
ok
end.
report_change() ->
io:format("Plugin configuration has changed. "
"Restart RabbitMQ for changes to take effect.~n"),
case os:type() of
{win32, _OsName} ->
io:format("If you have RabbitMQ running as a service then you must"
" reinstall by running~n rabbitmq-service.bat stop~n"
" rabbitmq-service.bat install~n"
" rabbitmq-service.bat start~n~n");
_ ->
ok
end.

273
src/rabbit_plugins_main.erl Normal file
View File

@ -0,0 +1,273 @@
%% 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) 2011-2012 VMware, Inc. All rights reserved.
%%
-module(rabbit_plugins_main).
-include("rabbit.hrl").
-export([start/0, stop/0]).
-define(VERBOSE_OPT, "-v").
-define(MINIMAL_OPT, "-m").
-define(ENABLED_OPT, "-E").
-define(ENABLED_ALL_OPT, "-e").
-define(VERBOSE_DEF, {?VERBOSE_OPT, flag}).
-define(MINIMAL_DEF, {?MINIMAL_OPT, flag}).
-define(ENABLED_DEF, {?ENABLED_OPT, flag}).
-define(ENABLED_ALL_DEF, {?ENABLED_ALL_OPT, flag}).
-define(GLOBAL_DEFS, []).
-define(COMMANDS,
[{list, [?VERBOSE_DEF, ?MINIMAL_DEF, ?ENABLED_DEF, ?ENABLED_ALL_DEF]},
enable,
disable]).
%%----------------------------------------------------------------------------
-ifdef(use_specs).
-spec(start/0 :: () -> no_return()).
-spec(stop/0 :: () -> 'ok').
-spec(usage/0 :: () -> no_return()).
-endif.
%%----------------------------------------------------------------------------
start() ->
{ok, [[PluginsFile|_]|_]} =
init:get_argument(enabled_plugins_file),
{ok, [[PluginsDir|_]|_]} = init:get_argument(plugins_dist_dir),
{Command, Opts, Args} =
case rabbit_misc:parse_arguments(?COMMANDS, ?GLOBAL_DEFS,
init:get_plain_arguments())
of
{ok, Res} -> Res;
no_command -> print_error("could not recognise command", []),
usage()
end,
PrintInvalidCommandError =
fun () ->
print_error("invalid command '~s'",
[string:join([atom_to_list(Command) | Args], " ")])
end,
case catch action(Command, Args, Opts, PluginsFile, PluginsDir) of
ok ->
rabbit_misc:quit(0);
{'EXIT', {function_clause, [{?MODULE, action, _} | _]}} ->
PrintInvalidCommandError(),
usage();
{'EXIT', {function_clause, [{?MODULE, action, _, _} | _]}} ->
PrintInvalidCommandError(),
usage();
{error, Reason} ->
print_error("~p", [Reason]),
rabbit_misc:quit(2);
{error_string, Reason} ->
print_error("~s", [Reason]),
rabbit_misc:quit(2);
Other ->
print_error("~p", [Other]),
rabbit_misc:quit(2)
end.
stop() ->
ok.
%%----------------------------------------------------------------------------
action(list, [], Opts, PluginsFile, PluginsDir) ->
action(list, [".*"], Opts, PluginsFile, PluginsDir);
action(list, [Pat], Opts, PluginsFile, PluginsDir) ->
format_plugins(Pat, Opts, PluginsFile, PluginsDir);
action(enable, ToEnable0, _Opts, PluginsFile, PluginsDir) ->
case ToEnable0 of
[] -> throw({error_string, "Not enough arguments for 'enable'"});
_ -> ok
end,
AllPlugins = rabbit_plugins:list(PluginsDir),
Enabled = rabbit_plugins:read_enabled(PluginsFile),
ImplicitlyEnabled = rabbit_plugins:dependencies(false,
Enabled, AllPlugins),
ToEnable = [list_to_atom(Name) || Name <- ToEnable0],
Missing = ToEnable -- plugin_names(AllPlugins),
case Missing of
[] -> ok;
_ -> throw({error_string,
fmt_list("The following plugins could not be found:",
Missing)})
end,
NewEnabled = lists:usort(Enabled ++ ToEnable),
write_enabled_plugins(PluginsFile, NewEnabled),
NewImplicitlyEnabled = rabbit_plugins:dependencies(false,
NewEnabled, AllPlugins),
maybe_warn_mochiweb(NewImplicitlyEnabled),
case NewEnabled -- ImplicitlyEnabled of
[] -> io:format("Plugin configuration unchanged.~n");
_ -> print_list("The following plugins have been enabled:",
NewImplicitlyEnabled -- ImplicitlyEnabled),
report_change()
end;
action(disable, ToDisable0, _Opts, PluginsFile, PluginsDir) ->
case ToDisable0 of
[] -> throw({error_string, "Not enough arguments for 'disable'"});
_ -> ok
end,
ToDisable = [list_to_atom(Name) || Name <- ToDisable0],
Enabled = rabbit_plugins:read_enabled(PluginsFile),
AllPlugins = rabbit_plugins:list(PluginsDir),
Missing = ToDisable -- plugin_names(AllPlugins),
case Missing of
[] -> ok;
_ -> print_list("Warning: the following plugins could not be found:",
Missing)
end,
ToDisableDeps = rabbit_plugins:dependencies(true, ToDisable, AllPlugins),
NewEnabled = Enabled -- ToDisableDeps,
case length(Enabled) =:= length(NewEnabled) of
true -> io:format("Plugin configuration unchanged.~n");
false -> ImplicitlyEnabled =
rabbit_plugins:dependencies(false, Enabled, AllPlugins),
NewImplicitlyEnabled =
rabbit_plugins:dependencies(false,
NewEnabled, AllPlugins),
print_list("The following plugins have been disabled:",
ImplicitlyEnabled -- NewImplicitlyEnabled),
write_enabled_plugins(PluginsFile, NewEnabled),
report_change()
end.
%%----------------------------------------------------------------------------
print_error(Format, Args) ->
rabbit_misc:format_stderr("Error: " ++ Format ++ "~n", Args).
usage() ->
io:format("~s", [rabbit_plugins_usage:usage()]),
rabbit_misc:quit(1).
%% Pretty print a list of plugins.
format_plugins(Pattern, Opts, PluginsFile, PluginsDir) ->
Verbose = proplists:get_bool(?VERBOSE_OPT, Opts),
Minimal = proplists:get_bool(?MINIMAL_OPT, Opts),
Format = case {Verbose, Minimal} of
{false, false} -> normal;
{true, false} -> verbose;
{false, true} -> minimal;
{true, true} -> throw({error_string,
"Cannot specify -m and -v together"})
end,
OnlyEnabled = proplists:get_bool(?ENABLED_OPT, Opts),
OnlyEnabledAll = proplists:get_bool(?ENABLED_ALL_OPT, Opts),
AvailablePlugins = rabbit_plugins:list(PluginsDir),
EnabledExplicitly = rabbit_plugins:read_enabled(PluginsFile),
EnabledImplicitly =
rabbit_plugins:dependencies(false, EnabledExplicitly,
AvailablePlugins) -- EnabledExplicitly,
{ok, RE} = re:compile(Pattern),
Plugins = [ Plugin ||
Plugin = #plugin{name = Name} <- AvailablePlugins,
re:run(atom_to_list(Name), RE, [{capture, none}]) =:= match,
if OnlyEnabled -> lists:member(Name, EnabledExplicitly);
OnlyEnabledAll -> (lists:member(Name,
EnabledExplicitly) or
lists:member(Name, EnabledImplicitly));
true -> true
end],
Plugins1 = usort_plugins(Plugins),
MaxWidth = lists:max([length(atom_to_list(Name)) ||
#plugin{name = Name} <- Plugins1] ++ [0]),
[format_plugin(P, EnabledExplicitly, EnabledImplicitly, Format,
MaxWidth) || P <- Plugins1],
ok.
format_plugin(#plugin{name = Name, version = Version,
description = Description, dependencies = Deps},
EnabledExplicitly, EnabledImplicitly, Format, MaxWidth) ->
Glyph = case {lists:member(Name, EnabledExplicitly),
lists:member(Name, EnabledImplicitly)} of
{true, false} -> "[E]";
{false, true} -> "[e]";
_ -> "[ ]"
end,
case Format of
minimal -> io:format("~s~n", [Name]);
normal -> io:format("~s ~-" ++ integer_to_list(MaxWidth) ++
"w ~s~n", [Glyph, Name, Version]);
verbose -> io:format("~s ~w~n", [Glyph, Name]),
io:format(" Version: \t~s~n", [Version]),
case Deps of
[] -> ok;
_ -> io:format(" Dependencies:\t~p~n", [Deps])
end,
io:format(" Description:\t~s~n", [Description]),
io:format("~n")
end.
print_list(Header, Plugins) ->
io:format(fmt_list(Header, Plugins)).
fmt_list(Header, Plugins) ->
lists:flatten(
[Header, $\n, [io_lib:format(" ~s~n", [P]) || P <- Plugins]]).
usort_plugins(Plugins) ->
lists:usort(fun plugins_cmp/2, Plugins).
plugins_cmp(#plugin{name = N1, version = V1},
#plugin{name = N2, version = V2}) ->
{N1, V1} =< {N2, V2}.
%% Return the names of the given plugins.
plugin_names(Plugins) ->
[Name || #plugin{name = Name} <- Plugins].
%% Write the enabled plugin names on disk.
write_enabled_plugins(PluginsFile, Plugins) ->
case rabbit_file:write_term_file(PluginsFile, [Plugins]) of
ok -> ok;
{error, Reason} -> throw({error, {cannot_write_enabled_plugins_file,
PluginsFile, Reason}})
end.
maybe_warn_mochiweb(Enabled) ->
V = erlang:system_info(otp_release),
case lists:member(mochiweb, Enabled) andalso V < "R13B01" of
true ->
Stars = string:copies("*", 80),
io:format("~n~n~s~n"
" Warning: Mochiweb enabled and Erlang version ~s "
"detected.~n"
" Enabling plugins that depend on Mochiweb is not "
"supported on this Erlang~n"
" version. At least R13B01 is required.~n~n"
" RabbitMQ will not start successfully in this "
"configuration. You *must*~n"
" disable the Mochiweb plugin, or upgrade Erlang.~n"
"~s~n~n~n", [Stars, V, Stars]);
false ->
ok
end.
report_change() ->
io:format("Plugin configuration has changed. "
"Restart RabbitMQ for changes to take effect.~n").

214
src/rabbit_prelaunch.erl
View File

@ -31,212 +31,21 @@
-spec(start/0 :: () -> no_return()). -spec(start/0 :: () -> no_return()).
-spec(stop/0 :: () -> 'ok'). -spec(stop/0 :: () -> 'ok').
%% Shut dialyzer up
-spec(terminate/1 :: (string()) -> no_return()).
-spec(terminate/2 :: (string(), [any()]) -> no_return()).
-endif. -endif.
%%---------------------------------------------------------------------------- %%----------------------------------------------------------------------------
start() -> start() ->
io:format("Activating RabbitMQ plugins ...~n"), [NodeStr] = init:get_plain_arguments(),
%% Determine our various directories
[EnabledPluginsFile, PluginsDistDir, UnpackedPluginDir, NodeStr] =
init:get_plain_arguments(),
RootName = UnpackedPluginDir ++ "/rabbit",
prepare_plugins(EnabledPluginsFile, PluginsDistDir, UnpackedPluginDir),
%% Build a list of required apps based on the fixed set, and any plugins
PluginApps = find_plugins(UnpackedPluginDir),
RequiredApps = ?BaseApps ++ PluginApps,
%% Build the entire set of dependencies - this will load the
%% applications along the way
AllApps = case catch sets:to_list(expand_dependencies(RequiredApps)) of
{failed_to_load_app, App, Err} ->
terminate("failed to load application ~s:~n~p",
[App, Err]);
AppList ->
AppList
end,
AppVersions = [determine_version(App) || App <- AllApps],
RabbitVersion = proplists:get_value(rabbit, AppVersions),
%% Build the overall release descriptor
RDesc = {release,
{"rabbit", RabbitVersion},
{erts, erlang:system_info(version)},
AppVersions},
%% Write it out to $RABBITMQ_PLUGINS_EXPAND_DIR/rabbit.rel
rabbit_file:write_file(RootName ++ ".rel", io_lib:format("~p.~n", [RDesc])),
%% We exclude mochiweb due to its optional use of fdsrv.
XRefExclude = [mochiweb],
%% Compile the script
ScriptFile = RootName ++ ".script",
case systools:make_script(RootName, [local, silent,
{exref, AllApps -- XRefExclude}]) of
{ok, Module, Warnings} ->
%% This gets lots of spurious no-source warnings when we
%% have .ez files, so we want to supress them to prevent
%% hiding real issues. On Ubuntu, we also get warnings
%% about kernel/stdlib sources being out of date, which we
%% also ignore for the same reason.
WarningStr = Module:format_warning(
[W || W <- Warnings,
case W of
{warning, {source_not_found, _}} -> false;
{warning, {obj_out_of_date, {_,_,WApp,_,_}}}
when WApp == mnesia;
WApp == stdlib;
WApp == kernel;
WApp == sasl;
WApp == crypto;
WApp == os_mon -> false;
_ -> true
end]),
case length(WarningStr) of
0 -> ok;
_ -> S = string:copies("*", 80),
io:format("~n~s~n~s~s~n~n", [S, WarningStr, S])
end,
ok;
{error, Module, Error} ->
terminate("generation of boot script file ~s failed:~n~s",
[ScriptFile, Module:format_error(Error)])
end,
case post_process_script(ScriptFile) of
ok -> ok;
{error, Reason} ->
terminate("post processing of boot script file ~s failed:~n~w",
[ScriptFile, Reason])
end,
case systools:script2boot(RootName) of
ok -> ok;
error -> terminate("failed to compile boot script file ~s",
[ScriptFile])
end,
io:format("~w plugins activated:~n", [length(PluginApps)]),
[io:format("* ~s-~s~n", [App, proplists:get_value(App, AppVersions)])
|| App <- PluginApps],
io:nl(),
ok = duplicate_node_check(NodeStr), ok = duplicate_node_check(NodeStr),
rabbit_misc:quit(0),
terminate(0),
ok. ok.
stop() -> stop() ->
ok. ok.
determine_version(App) -> %%----------------------------------------------------------------------------
application:load(App),
{ok, Vsn} = application:get_key(App, vsn),
{App, Vsn}.
delete_recursively(Fn) ->
case rabbit_file:recursive_delete([Fn]) of
ok -> ok;
{error, {Path, E}} -> {error, {cannot_delete, Path, E}};
Error -> Error
end.
prepare_plugins(EnabledPluginsFile, PluginsDistDir, DestDir) ->
AllPlugins = rabbit_plugins:find_plugins(PluginsDistDir),
Enabled = rabbit_plugins:read_enabled_plugins(EnabledPluginsFile),
ToUnpack = rabbit_plugins:calculate_required_plugins(Enabled, AllPlugins),
ToUnpackPlugins = rabbit_plugins:lookup_plugins(ToUnpack, AllPlugins),
Missing = Enabled -- rabbit_plugins:plugin_names(ToUnpackPlugins),
case Missing of
[] -> ok;
_ -> io:format("Warning: the following enabled plugins were "
"not found: ~p~n", [Missing])
end,
%% Eliminate the contents of the destination directory
case delete_recursively(DestDir) of
ok -> ok;
{error, E} -> terminate("Could not delete dir ~s (~p)", [DestDir, E])
end,
case filelib:ensure_dir(DestDir ++ "/") of
ok -> ok;
{error, E2} -> terminate("Could not create dir ~s (~p)", [DestDir, E2])
end,
[prepare_plugin(Plugin, DestDir) || Plugin <- ToUnpackPlugins].
prepare_plugin(#plugin{type = ez, location = Location}, PluginDestDir) ->
zip:unzip(Location, [{cwd, PluginDestDir}]);
prepare_plugin(#plugin{type = dir, name = Name, location = Location},
PluginsDestDir) ->
rabbit_file:recursive_copy(Location,
filename:join([PluginsDestDir, Name])).
find_plugins(PluginDir) ->
[prepare_dir_plugin(PluginName) ||
PluginName <- filelib:wildcard(PluginDir ++ "/*/ebin/*.app")].
prepare_dir_plugin(PluginAppDescFn) ->
%% Add the plugin ebin directory to the load path
PluginEBinDirN = filename:dirname(PluginAppDescFn),
code:add_path(PluginEBinDirN),
%% We want the second-last token
NameTokens = string:tokens(PluginAppDescFn,"/."),
PluginNameString = lists:nth(length(NameTokens) - 1, NameTokens),
list_to_atom(PluginNameString).
expand_dependencies(Pending) ->
expand_dependencies(sets:new(), Pending).
expand_dependencies(Current, []) ->
Current;
expand_dependencies(Current, [Next|Rest]) ->
case sets:is_element(Next, Current) of
true ->
expand_dependencies(Current, Rest);
false ->
case application:load(Next) of
ok ->
ok;
{error, {already_loaded, _}} ->
ok;
{error, Reason} ->
throw({failed_to_load_app, Next, Reason})
end,
{ok, Required} = application:get_key(Next, applications),
Unique = [A || A <- Required, not(sets:is_element(A, Current))],
expand_dependencies(sets:add_element(Next, Current), Rest ++ Unique)
end.
post_process_script(ScriptFile) ->
case file:consult(ScriptFile) of
{ok, [{script, Name, Entries}]} ->
NewEntries = lists:flatmap(fun process_entry/1, Entries),
case file:open(ScriptFile, [write]) of
{ok, Fd} ->
io:format(Fd, "%% script generated at ~w ~w~n~p.~n",
[date(), time(), {script, Name, NewEntries}]),
file:close(Fd),
ok;
{error, OReason} ->
{error, {failed_to_open_script_file_for_writing, OReason}}
end;
{error, Reason} ->
{error, {failed_to_load_script, Reason}}
end.
process_entry(Entry = {apply,{application,start_boot,[mnesia,permanent]}}) ->
[{apply,{rabbit,maybe_hipe_compile,[]}},
{apply,{rabbit,prepare,[]}}, Entry];
process_entry(Entry) ->
[Entry].
%% Check whether a node with the same name is already running %% Check whether a node with the same name is already running
duplicate_node_check([]) -> duplicate_node_check([]) ->
@ -252,11 +61,11 @@ duplicate_node_check(NodeStr) ->
"already running on ~p~n", "already running on ~p~n",
[NodeName, NodeHost]), [NodeName, NodeHost]),
io:format(rabbit_nodes:diagnostics([Node]) ++ "~n"), io:format(rabbit_nodes:diagnostics([Node]) ++ "~n"),
terminate(?ERROR_CODE); rabbit_misc:quit(?ERROR_CODE);
false -> ok false -> ok
end; end;
{error, EpmdReason} -> {error, EpmdReason} ->
terminate("epmd error for host ~p: ~p (~s)~n", rabbit_misc:quit("epmd error for host ~p: ~p (~s)~n",
[NodeHost, EpmdReason, [NodeHost, EpmdReason,
case EpmdReason of case EpmdReason of
address -> "unable to establish tcp connection"; address -> "unable to establish tcp connection";
@ -264,16 +73,3 @@ duplicate_node_check(NodeStr) ->
_ -> inet:format_error(EpmdReason) _ -> inet:format_error(EpmdReason)
end]) end])
end. end.
terminate(Fmt, Args) ->
io:format("ERROR: " ++ Fmt ++ "~n", Args),
terminate(?ERROR_CODE).
terminate(Status) ->
case os:type() of
{unix, _} -> halt(Status);
{win32, _} -> init:stop(Status),
receive
after infinity -> ok
end
end.

15
src/rabbit_reader.erl
View File

@ -25,7 +25,7 @@
-export([init/4, mainloop/2]). -export([init/4, mainloop/2]).
-export([conserve_resources/2, server_properties/1]). -export([conserve_resources/3, server_properties/1]).
-define(HANDSHAKE_TIMEOUT, 10). -define(HANDSHAKE_TIMEOUT, 10).
-define(NORMAL_TIMEOUT, 3). -define(NORMAL_TIMEOUT, 3).
@ -71,7 +71,7 @@
-spec(info/2 :: (pid(), rabbit_types:info_keys()) -> rabbit_types:infos()). -spec(info/2 :: (pid(), rabbit_types:info_keys()) -> rabbit_types:infos()).
-spec(force_event_refresh/1 :: (pid()) -> 'ok'). -spec(force_event_refresh/1 :: (pid()) -> 'ok').
-spec(shutdown/2 :: (pid(), string()) -> 'ok'). -spec(shutdown/2 :: (pid(), string()) -> 'ok').
-spec(conserve_resources/2 :: (pid(), boolean()) -> 'ok'). -spec(conserve_resources/3 :: (pid(), atom(), boolean()) -> 'ok').
-spec(server_properties/1 :: (rabbit_types:protocol()) -> -spec(server_properties/1 :: (rabbit_types:protocol()) ->
rabbit_framing:amqp_table()). rabbit_framing:amqp_table()).
@ -133,7 +133,7 @@ info(Pid, Items) ->
force_event_refresh(Pid) -> force_event_refresh(Pid) ->
gen_server:cast(Pid, force_event_refresh). gen_server:cast(Pid, force_event_refresh).
conserve_resources(Pid, Conserve) -> conserve_resources(Pid, _Source, Conserve) ->
Pid ! {conserve_resources, Conserve}, Pid ! {conserve_resources, Conserve},
ok. ok.
@ -222,14 +222,7 @@ start_connection(Parent, ChannelSupSupPid, Collector, StartHeartbeatFun, Deb,
last_blocked_by = none, last_blocked_by = none,
last_blocked_at = never}, last_blocked_at = never},
try try
BufSizes = inet_op(fun () -> ok = inet_op(fun () -> rabbit_net:tune_buffer_size(ClientSock) end),
rabbit_net:getopts(
ClientSock, [sndbuf, recbuf, buffer])
end),
BufSz = lists:max([Sz || {_Opt, Sz} <- BufSizes]),
ok = inet_op(fun () ->
rabbit_net:setopts(ClientSock, [{buffer, BufSz}])
end),
recvloop(Deb, switch_callback(rabbit_event:init_stats_timer( recvloop(Deb, switch_callback(rabbit_event:init_stats_timer(
State, #v1.stats_timer), State, #v1.stats_timer),
handshake, 8)), handshake, 8)),

38
src/rabbit_sup.erl
View File

@ -19,6 +19,8 @@
-behaviour(supervisor). -behaviour(supervisor).
-export([start_link/0, start_child/1, start_child/2, start_child/3, -export([start_link/0, start_child/1, start_child/2, start_child/3,
start_supervisor_child/1, start_supervisor_child/2,
start_supervisor_child/3,
start_restartable_child/1, start_restartable_child/2, stop_child/1]). start_restartable_child/1, start_restartable_child/2, stop_child/1]).
-export([init/1]). -export([init/1]).
@ -33,7 +35,11 @@
-spec(start_link/0 :: () -> rabbit_types:ok_pid_or_error()). -spec(start_link/0 :: () -> rabbit_types:ok_pid_or_error()).
-spec(start_child/1 :: (atom()) -> 'ok'). -spec(start_child/1 :: (atom()) -> 'ok').
-spec(start_child/2 :: (atom(), [any()]) -> 'ok').
-spec(start_child/3 :: (atom(), atom(), [any()]) -> 'ok'). -spec(start_child/3 :: (atom(), atom(), [any()]) -> 'ok').
-spec(start_supervisor_child/1 :: (atom()) -> 'ok').
-spec(start_supervisor_child/2 :: (atom(), [any()]) -> 'ok').
-spec(start_supervisor_child/3 :: (atom(), atom(), [any()]) -> 'ok').
-spec(start_restartable_child/1 :: (atom()) -> 'ok'). -spec(start_restartable_child/1 :: (atom()) -> 'ok').
-spec(start_restartable_child/2 :: (atom(), [any()]) -> 'ok'). -spec(start_restartable_child/2 :: (atom(), [any()]) -> 'ok').
-spec(stop_child/1 :: (atom()) -> rabbit_types:ok_or_error(any())). -spec(stop_child/1 :: (atom()) -> rabbit_types:ok_or_error(any())).
@ -42,22 +48,29 @@
%%---------------------------------------------------------------------------- %%----------------------------------------------------------------------------
start_link() -> start_link() -> supervisor:start_link({local, ?SERVER}, ?MODULE, []).
supervisor:start_link({local, ?SERVER}, ?MODULE, []).
start_child(Mod) -> start_child(Mod) -> start_child(Mod, []).
start_child(Mod, []).
start_child(Mod, Args) -> start_child(Mod, Args) -> start_child(Mod, Mod, Args).
start_child(Mod, Mod, Args).
start_child(ChildId, Mod, Args) -> start_child(ChildId, Mod, Args) ->
child_reply(supervisor:start_child(?SERVER, child_reply(supervisor:start_child(
{ChildId, {Mod, start_link, Args}, ?SERVER,
transient, ?MAX_WAIT, worker, [Mod]})). {ChildId, {Mod, start_link, Args},
transient, ?MAX_WAIT, worker, [Mod]})).
start_restartable_child(Mod) -> start_supervisor_child(Mod) -> start_supervisor_child(Mod, []).
start_restartable_child(Mod, []).
start_supervisor_child(Mod, Args) -> start_supervisor_child(Mod, Mod, Args).
start_supervisor_child(ChildId, Mod, Args) ->
child_reply(supervisor:start_child(
?SERVER,
{ChildId, {Mod, start_link, Args},
transient, infinity, supervisor, [Mod]})).
start_restartable_child(Mod) -> start_restartable_child(Mod, []).
start_restartable_child(Mod, Args) -> start_restartable_child(Mod, Args) ->
Name = list_to_atom(atom_to_list(Mod) ++ "_sup"), Name = list_to_atom(atom_to_list(Mod) ++ "_sup"),
@ -73,8 +86,7 @@ stop_child(ChildId) ->
E -> E E -> E
end. end.
init([]) -> init([]) -> {ok, {{one_for_all, 0, 1}, []}}.
{ok, {{one_for_all, 0, 1}, []}}.
%%---------------------------------------------------------------------------- %%----------------------------------------------------------------------------

81
src/rabbit_tests.erl
View File

@ -50,7 +50,7 @@ all_tests() ->
passed = test_app_management(), passed = test_app_management(),
passed = test_log_management_during_startup(), passed = test_log_management_during_startup(),
passed = test_statistics(), passed = test_statistics(),
passed = test_option_parser(), passed = test_arguments_parser(),
passed = test_cluster_management(), passed = test_cluster_management(),
passed = test_user_management(), passed = test_user_management(),
passed = test_runtime_parameters(), passed = test_runtime_parameters(),
@ -801,27 +801,57 @@ test_log_management_during_startup() ->
ok = control_action(start_app, []), ok = control_action(start_app, []),
passed. passed.
test_option_parser() -> test_arguments_parser() ->
%% command and arguments should just pass through GlobalOpts1 = [{"-f1", flag}, {"-o1", {option, "foo"}}],
ok = check_get_options({["mock_command", "arg1", "arg2"], []}, Commands1 = [command1, {command2, [{"-f2", flag}, {"-o2", {option, "bar"}}]}],
[], ["mock_command", "arg1", "arg2"]),
%% get flags GetOptions =
ok = check_get_options( fun (Args) ->
{["mock_command", "arg1"], [{"-f", true}, {"-f2", false}]}, rabbit_misc:parse_arguments(Commands1, GlobalOpts1, Args)
[{flag, "-f"}, {flag, "-f2"}], ["mock_command", "arg1", "-f"]), end,
%% get options check_parse_arguments(no_command, GetOptions, []),
ok = check_get_options( check_parse_arguments(no_command, GetOptions, ["foo", "bar"]),
{["mock_command"], [{"-foo", "bar"}, {"-baz", "notbaz"}]}, check_parse_arguments(
[{option, "-foo", "notfoo"}, {option, "-baz", "notbaz"}], {ok, {command1, [{"-f1", false}, {"-o1", "foo"}], []}},
["mock_command", "-foo", "bar"]), GetOptions, ["command1"]),
check_parse_arguments(
{ok, {command1, [{"-f1", false}, {"-o1", "blah"}], []}},
GetOptions, ["command1", "-o1", "blah"]),
check_parse_arguments(
{ok, {command1, [{"-f1", true}, {"-o1", "foo"}], []}},
GetOptions, ["command1", "-f1"]),
check_parse_arguments(
{ok, {command1, [{"-f1", false}, {"-o1", "blah"}], []}},
GetOptions, ["-o1", "blah", "command1"]),
check_parse_arguments(
{ok, {command1, [{"-f1", false}, {"-o1", "blah"}], ["quux"]}},
GetOptions, ["-o1", "blah", "command1", "quux"]),
check_parse_arguments(
{ok, {command1, [{"-f1", true}, {"-o1", "blah"}], ["quux", "baz"]}},
GetOptions, ["command1", "quux", "-f1", "-o1", "blah", "baz"]),
%% For duplicate flags, the last one counts
check_parse_arguments(
{ok, {command1, [{"-f1", false}, {"-o1", "second"}], []}},
GetOptions, ["-o1", "first", "command1", "-o1", "second"]),
%% If the flag "eats" the command, the command won't be recognised
check_parse_arguments(no_command, GetOptions,
["-o1", "command1", "quux"]),
%% If a flag eats another flag, the eaten flag won't be recognised
check_parse_arguments(
{ok, {command1, [{"-f1", false}, {"-o1", "-f1"}], []}},
GetOptions, ["command1", "-o1", "-f1"]),
%% shuffled and interleaved arguments and options %% Now for some command-specific flags...
ok = check_get_options( check_parse_arguments(
{["a1", "a2", "a3"], [{"-o1", "hello"}, {"-o2", "noto2"}, {"-f", true}]}, {ok, {command2, [{"-f1", false}, {"-f2", false},
[{option, "-o1", "noto1"}, {flag, "-f"}, {option, "-o2", "noto2"}], {"-o1", "foo"}, {"-o2", "bar"}], []}},
["-f", "a1", "-o1", "hello", "a2", "a3"]), GetOptions, ["command2"]),
check_parse_arguments(
{ok, {command2, [{"-f1", false}, {"-f2", true},
{"-o1", "baz"}, {"-o2", "bar"}], ["quux", "foo"]}},
GetOptions, ["-f2", "command2", "quux", "-o1", "baz", "foo"]),
passed. passed.
@ -1573,7 +1603,7 @@ control_action(Command, Args, NewOpts) ->
expand_options(default_options(), NewOpts)). expand_options(default_options(), NewOpts)).
control_action(Command, Node, Args, Opts) -> control_action(Command, Node, Args, Opts) ->
case catch rabbit_control:action( case catch rabbit_control_main:action(
Command, Node, Args, Opts, Command, Node, Args, Opts,
fun (Format, Args1) -> fun (Format, Args1) ->
io:format(Format ++ " ...~n", Args1) io:format(Format ++ " ...~n", Args1)
@ -1605,10 +1635,13 @@ expand_options(As, Bs) ->
end end
end, Bs, As). end, Bs, As).
check_get_options({ExpArgs, ExpOpts}, Defs, Args) -> check_parse_arguments(ExpRes, Fun, As) ->
{ExpArgs, ResOpts} = rabbit_misc:get_options(Defs, Args), SortRes =
true = lists:sort(ExpOpts) == lists:sort(ResOpts), % don't care about the order fun (no_command) -> no_command;
ok. ({ok, {C, KVs, As1}}) -> {ok, {C, lists:sort(KVs), As1}}
end,
true = SortRes(ExpRes) =:= SortRes(Fun(As)).
empty_files(Files) -> empty_files(Files) ->
[case file:read_file_info(File) of [case file:read_file_info(File) of

28
src/rabbit_variable_queue.erl
View File

@ -323,7 +323,6 @@
-type(timestamp() :: {non_neg_integer(), non_neg_integer(), non_neg_integer()}). -type(timestamp() :: {non_neg_integer(), non_neg_integer(), non_neg_integer()}).
-type(seq_id() :: non_neg_integer()). -type(seq_id() :: non_neg_integer()).
-type(ack() :: seq_id()).
-type(rates() :: #rates { egress :: {timestamp(), non_neg_integer()}, -type(rates() :: #rates { egress :: {timestamp(), non_neg_integer()},
ingress :: {timestamp(), non_neg_integer()}, ingress :: {timestamp(), non_neg_integer()},
@ -335,6 +334,13 @@
count :: non_neg_integer(), count :: non_neg_integer(),
end_seq_id :: non_neg_integer() }). end_seq_id :: non_neg_integer() }).
%% The compiler (rightfully) complains that ack() and state() are
%% unused. For this reason we duplicate a -spec from
%% rabbit_backing_queue with the only intent being to remove
%% warnings. The problem here is that we can't parameterise the BQ
%% behaviour by these two types as we would like to. We still leave
%% these here for documentation purposes.
-type(ack() :: seq_id()).
-type(state() :: #vqstate { -type(state() :: #vqstate {
q1 :: ?QUEUE:?QUEUE(), q1 :: ?QUEUE:?QUEUE(),
q2 :: ?QUEUE:?QUEUE(), q2 :: ?QUEUE:?QUEUE(),
@ -368,6 +374,8 @@
ack_out_counter :: non_neg_integer(), ack_out_counter :: non_neg_integer(),
ack_in_counter :: non_neg_integer(), ack_in_counter :: non_neg_integer(),
ack_rates :: rates() }). ack_rates :: rates() }).
%% Duplicated from rabbit_backing_queue
-spec(ack/2 :: ([ack()], state()) -> {[rabbit_guid:guid()], state()}).
-spec(multiple_routing_keys/0 :: () -> 'ok'). -spec(multiple_routing_keys/0 :: () -> 'ok').
@ -1458,16 +1466,14 @@ reduce_memory_use(AlphaBetaFun, BetaDeltaFun, AckFun,
%% determined based on which is growing faster. Whichever %% determined based on which is growing faster. Whichever
%% comes second may very well get a quota of 0 if the %% comes second may very well get a quota of 0 if the
%% first manages to push out the max number of messages. %% first manages to push out the max number of messages.
S1 -> {_, State2} = S1 -> Funs = case ((AvgAckIngress - AvgAckEgress) >
lists:foldl(fun (ReduceFun, {QuotaN, StateN}) -> (AvgIngress - AvgEgress)) of
ReduceFun(QuotaN, StateN) true -> [AckFun, AlphaBetaFun];
end, false -> [AlphaBetaFun, AckFun]
{S1, State}, end,
case (AvgAckIngress - AvgAckEgress) > {_, State2} = lists:foldl(fun (ReduceFun, {QuotaN, StateN}) ->
(AvgIngress - AvgEgress) of ReduceFun(QuotaN, StateN)
true -> [AckFun, AlphaBetaFun]; end, {S1, State}, Funs),
false -> [AlphaBetaFun, AckFun]
end),
{true, State2} {true, State2}
end, end,