summaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/rabbit_access_control.erl7
-rw-r--r--src/rabbit_amqqueue.erl9
-rw-r--r--src/rabbit_amqqueue_process.erl6
-rw-r--r--src/rabbit_backing_queue.erl17
-rw-r--r--src/rabbit_control.erl6
-rw-r--r--src/rabbit_invariable_queue.erl8
-rw-r--r--src/rabbit_mnesia.erl98
-rw-r--r--src/rabbit_sup.erl8
-rw-r--r--src/rabbit_tests.erl38
9 files changed, 127 insertions, 70 deletions
diff --git a/src/rabbit_access_control.erl b/src/rabbit_access_control.erl
index 7d1839bb6e..30bae25e5a 100644
--- a/src/rabbit_access_control.erl
+++ b/src/rabbit_access_control.erl
@@ -172,9 +172,14 @@ check_resource_access(Username,
[] ->
false;
[#user_permission{permission = P}] ->
+ PermRegexp = case element(permission_index(Permission), P) of
+ %% <<"^$">> breaks Emacs' erlang mode
+ <<"">> -> <<$^, $$>>;
+ RE -> RE
+ end,
case regexp:match(
binary_to_list(Name),
- binary_to_list(element(permission_index(Permission), P))) of
+ binary_to_list(PermRegexp)) of
{match, _, _} -> true;
nomatch -> false
end
diff --git a/src/rabbit_amqqueue.erl b/src/rabbit_amqqueue.erl
index f1b527681c..e5faef5416 100644
--- a/src/rabbit_amqqueue.erl
+++ b/src/rabbit_amqqueue.erl
@@ -31,7 +31,7 @@
-module(rabbit_amqqueue).
--export([start/0, declare/5, delete/3, purge/1]).
+-export([start/0, stop/0, declare/5, delete/3, purge/1]).
-export([internal_declare/2, internal_delete/1,
maybe_run_queue_via_backing_queue/2,
update_ram_duration/1, set_ram_duration_target/2,
@@ -71,6 +71,7 @@
'ok' | {'error', [{'error' | 'exit' | 'throw', any()}]}).
-spec(start/0 :: () -> 'ok').
+-spec(stop/0 :: () -> 'ok').
-spec(declare/5 ::
(name(), boolean(), boolean(),
rabbit_framing:amqp_table(), rabbit_types:maybe(pid()))
@@ -154,7 +155,7 @@
start() ->
DurableQueues = find_durable_queues(),
- {ok, BQ} = application:get_env(backing_queue_module),
+ {ok, BQ} = application:get_env(rabbit, backing_queue_module),
ok = BQ:start([QName || #amqqueue{name = QName} <- DurableQueues]),
{ok,_} = supervisor:start_child(
rabbit_sup,
@@ -164,6 +165,10 @@ start() ->
_RealDurableQueues = recover_durable_queues(DurableQueues),
ok.
+stop() ->
+ ok = supervisor:terminate_child(rabbit_sup, rabbit_amqqueue_sup),
+ ok = supervisor:delete_child(rabbit_sup, rabbit_amqqueue_sup).
+
find_durable_queues() ->
Node = node(),
%% TODO: use dirty ops instead
diff --git a/src/rabbit_amqqueue_process.erl b/src/rabbit_amqqueue_process.erl
index 2fb60e9675..a2cbcf5517 100644
--- a/src/rabbit_amqqueue_process.erl
+++ b/src/rabbit_amqqueue_process.erl
@@ -179,7 +179,7 @@ noreply(NewState) ->
next_state(State) ->
State1 = #q{backing_queue = BQ, backing_queue_state = BQS} =
ensure_rate_timer(State),
- case BQ:needs_sync(BQS)of
+ case BQ:needs_idle_timeout(BQS)of
true -> {ensure_sync_timer(State1), 0};
false -> {stop_sync_timer(State1), hibernate}
end.
@@ -188,7 +188,7 @@ ensure_sync_timer(State = #q{sync_timer_ref = undefined, backing_queue = BQ}) ->
{ok, TRef} = timer:apply_after(
?SYNC_INTERVAL,
rabbit_amqqueue, maybe_run_queue_via_backing_queue,
- [self(), fun (BQS) -> BQ:sync(BQS) end]),
+ [self(), fun (BQS) -> BQ:idle_timeout(BQS) end]),
State#q{sync_timer_ref = TRef};
ensure_sync_timer(State) ->
State.
@@ -822,7 +822,7 @@ handle_info({'DOWN', _MonitorRef, process, DownPid, _Reason}, State) ->
handle_info(timeout, State = #q{backing_queue = BQ}) ->
noreply(maybe_run_queue_via_backing_queue(
- fun (BQS) -> BQ:sync(BQS) end, State));
+ fun (BQS) -> BQ:idle_timeout(BQS) end, State));
handle_info({'EXIT', _Pid, Reason}, State) ->
{stop, Reason, State};
diff --git a/src/rabbit_backing_queue.erl b/src/rabbit_backing_queue.erl
index 432d62900b..b76ae11ebc 100644
--- a/src/rabbit_backing_queue.erl
+++ b/src/rabbit_backing_queue.erl
@@ -113,14 +113,15 @@ behaviour_info(callbacks) ->
%% queue.
{ram_duration, 1},
- %% Should 'sync' be called as soon as the queue process can
- %% manage (either on an empty mailbox, or when a timer fires)?
- {needs_sync, 1},
-
- %% Called (eventually) after needs_sync returns 'true'. Note this
- %% may be called more than once for each 'true' returned from
- %% needs_sync.
- {sync, 1},
+ %% Should 'idle_timeout' be called as soon as the queue process
+ %% can manage (either on an empty mailbox, or when a timer
+ %% fires)?
+ {needs_idle_timeout, 1},
+
+ %% Called (eventually) after needs_idle_timeout returns
+ %% 'true'. Note this may be called more than once for each 'true'
+ %% returned from needs_idle_timeout.
+ {idle_timeout, 1},
%% Called immediately before the queue hibernates.
{handle_pre_hibernate, 1},
diff --git a/src/rabbit_control.erl b/src/rabbit_control.erl
index 95a49f8679..6e6ad06cb3 100644
--- a/src/rabbit_control.erl
+++ b/src/rabbit_control.erl
@@ -160,6 +160,12 @@ action(cluster, Node, ClusterNodeSs, Inform) ->
[Node, ClusterNodes]),
rpc_call(Node, rabbit_mnesia, cluster, [ClusterNodes]);
+action(force_cluster, Node, ClusterNodeSs, Inform) ->
+ ClusterNodes = lists:map(fun list_to_atom/1, ClusterNodeSs),
+ Inform("Forcefully clustering node ~p with ~p (ignoring offline nodes)",
+ [Node, ClusterNodes]),
+ rpc_call(Node, rabbit_mnesia, force_cluster, [ClusterNodes]);
+
action(status, Node, [], Inform) ->
Inform("Status of node ~p", [Node]),
case call(Node, {rabbit, status, []}) of
diff --git a/src/rabbit_invariable_queue.erl b/src/rabbit_invariable_queue.erl
index 8214b976c4..cf8bc8f77e 100644
--- a/src/rabbit_invariable_queue.erl
+++ b/src/rabbit_invariable_queue.erl
@@ -34,8 +34,8 @@
-export([init/3, terminate/1, delete_and_terminate/1, purge/1, publish/2,
publish_delivered/3, fetch/2, ack/2, tx_publish/3, tx_ack/3,
tx_rollback/2, tx_commit/3, requeue/2, len/1, is_empty/1,
- set_ram_duration_target/2, ram_duration/1, needs_sync/1, sync/1,
- handle_pre_hibernate/1, status/1]).
+ set_ram_duration_target/2, ram_duration/1, needs_idle_timeout/1,
+ idle_timeout/1, handle_pre_hibernate/1, status/1]).
-export([start/1]).
@@ -197,9 +197,9 @@ set_ram_duration_target(_DurationTarget, State) -> State.
ram_duration(State) -> {0, State}.
-needs_sync(_State) -> false.
+needs_idle_timeout(_State) -> false.
-sync(State) -> State.
+idle_timeout(State) -> State.
handle_pre_hibernate(State) -> State.
diff --git a/src/rabbit_mnesia.erl b/src/rabbit_mnesia.erl
index 5c14ba7b16..e2b6927f8c 100644
--- a/src/rabbit_mnesia.erl
+++ b/src/rabbit_mnesia.erl
@@ -33,8 +33,8 @@
-module(rabbit_mnesia).
-export([ensure_mnesia_dir/0, dir/0, status/0, init/0, is_db_empty/0,
- cluster/1, reset/0, force_reset/0, is_clustered/0,
- empty_ram_only_tables/0]).
+ cluster/1, force_cluster/1, reset/0, force_reset/0,
+ is_clustered/0, empty_ram_only_tables/0]).
-export([table_names/0]).
@@ -58,6 +58,8 @@
-spec(init/0 :: () -> 'ok').
-spec(is_db_empty/0 :: () -> boolean()).
-spec(cluster/1 :: ([node()]) -> 'ok').
+-spec(force_cluster/1 :: ([node()]) -> 'ok').
+-spec(cluster/2 :: ([node()], boolean()) -> 'ok').
-spec(reset/0 :: () -> 'ok').
-spec(force_reset/0 :: () -> 'ok').
-spec(is_clustered/0 :: () -> boolean()).
@@ -88,7 +90,7 @@ status() ->
init() ->
ok = ensure_mnesia_running(),
ok = ensure_mnesia_dir(),
- ok = init_db(read_cluster_nodes_config()),
+ ok = init_db(read_cluster_nodes_config(), true),
ok = wait_for_tables(),
ok.
@@ -96,16 +98,22 @@ is_db_empty() ->
lists:all(fun (Tab) -> mnesia:dirty_first(Tab) == '$end_of_table' end,
table_names()).
+cluster(ClusterNodes) ->
+ cluster(ClusterNodes, false).
+force_cluster(ClusterNodes) ->
+ cluster(ClusterNodes, true).
+
%% Alter which disk nodes this node is clustered with. This can be a
%% subset of all the disk nodes in the cluster but can (and should)
%% include the node itself if it is to be a disk rather than a ram
-%% node.
-cluster(ClusterNodes) ->
+%% node. If Force is false, only connections to online nodes are
+%% allowed.
+cluster(ClusterNodes, Force) ->
ok = ensure_mnesia_not_running(),
ok = ensure_mnesia_dir(),
rabbit_misc:ensure_ok(mnesia:start(), cannot_start_mnesia),
try
- ok = init_db(ClusterNodes),
+ ok = init_db(ClusterNodes, Force),
ok = wait_for_tables(),
ok = create_cluster_nodes_config(ClusterNodes)
after
@@ -277,38 +285,56 @@ delete_cluster_nodes_config() ->
%% Take a cluster node config and create the right kind of node - a
%% standalone disk node, or disk or ram node connected to the
-%% specified cluster nodes.
-init_db(ClusterNodes) ->
- case mnesia:change_config(extra_db_nodes, ClusterNodes -- [node()]) of
- {ok, []} ->
- case mnesia:system_info(use_dir) of
- true ->
- case check_schema_integrity() of
- ok ->
- ok;
- {error, Reason} ->
- %% NB: we cannot use rabbit_log here since
- %% it may not have been started yet
- error_logger:warning_msg(
- "schema integrity check failed: ~p~n"
- "moving database to backup location "
- "and recreating schema from scratch~n",
- [Reason]),
- ok = move_db(),
+%% specified cluster nodes. If Force is false, don't allow
+%% connections to offline nodes.
+init_db(ClusterNodes, Force) ->
+ UClusterNodes = lists:usort(ClusterNodes),
+ ProperClusterNodes = UClusterNodes -- [node()],
+ case mnesia:change_config(extra_db_nodes, ProperClusterNodes) of
+ {ok, Nodes} ->
+ case Force of
+ false ->
+ FailedClusterNodes = ProperClusterNodes -- Nodes,
+ case FailedClusterNodes of
+ [] -> ok;
+ _ ->
+ throw({error, {failed_to_cluster_with,
+ FailedClusterNodes,
+ "Mnesia could not connect to some nodes."}})
+ end;
+ _ -> ok
+ end,
+ case Nodes of
+ [] ->
+ case mnesia:system_info(use_dir) of
+ true ->
+ case check_schema_integrity() of
+ ok ->
+ ok;
+ {error, Reason} ->
+ %% NB: we cannot use rabbit_log here since
+ %% it may not have been started yet
+ error_logger:warning_msg(
+ "schema integrity check failed: ~p~n"
+ "moving database to backup location "
+ "and recreating schema from scratch~n",
+ [Reason]),
+ ok = move_db(),
+ ok = create_schema()
+ end;
+ false ->
ok = create_schema()
end;
- false ->
- ok = create_schema()
- end;
- {ok, [_|_]} ->
- IsDiskNode = ClusterNodes == [] orelse
- lists:member(node(), ClusterNodes),
- ok = wait_for_replicated_tables(),
- ok = create_local_table_copy(schema, disc_copies),
- ok = create_local_table_copies(case IsDiskNode of
- true -> disc;
- false -> ram
- end);
+ [_|_] ->
+ IsDiskNode = ClusterNodes == [] orelse
+ lists:member(node(), ClusterNodes),
+ ok = wait_for_replicated_tables(),
+ ok = create_local_table_copy(schema, disc_copies),
+ ok = create_local_table_copies(case IsDiskNode of
+ true -> disc;
+ false -> ram
+ end)
+ end;
{error, Reason} ->
%% one reason we may end up here is if we try to join
%% nodes together that are currently running standalone or
diff --git a/src/rabbit_sup.erl b/src/rabbit_sup.erl
index 2c5e51125e..97613d17a5 100644
--- a/src/rabbit_sup.erl
+++ b/src/rabbit_sup.erl
@@ -34,7 +34,7 @@
-behaviour(supervisor).
-export([start_link/0, start_child/1, start_child/2, start_child/3,
- start_restartable_child/1, start_restartable_child/2]).
+ start_restartable_child/1, start_restartable_child/2, stop_child/1]).
-export([init/1]).
@@ -69,5 +69,11 @@ start_restartable_child(Mod, Args) ->
transient, infinity, supervisor, [rabbit_restartable_sup]}),
ok.
+stop_child(ChildId) ->
+ case supervisor:terminate_child(?SERVER, ChildId) of
+ ok -> supervisor:delete_child(?SERVER, ChildId);
+ E -> E
+ end.
+
init([]) ->
{ok, {{one_for_all, 0, 1}, []}}.
diff --git a/src/rabbit_tests.erl b/src/rabbit_tests.erl
index eaa195e646..d2c71b6b19 100644
--- a/src/rabbit_tests.erl
+++ b/src/rabbit_tests.erl
@@ -599,19 +599,19 @@ test_cluster_management() ->
ok = control_action(reset, []),
lists:foreach(fun (Arg) ->
- ok = control_action(cluster, Arg),
+ ok = control_action(force_cluster, Arg),
ok
end,
ClusteringSequence),
lists:foreach(fun (Arg) ->
ok = control_action(reset, []),
- ok = control_action(cluster, Arg),
+ ok = control_action(force_cluster, Arg),
ok
end,
ClusteringSequence),
ok = control_action(reset, []),
lists:foreach(fun (Arg) ->
- ok = control_action(cluster, Arg),
+ ok = control_action(force_cluster, Arg),
ok = control_action(start_app, []),
ok = control_action(stop_app, []),
ok
@@ -619,7 +619,7 @@ test_cluster_management() ->
ClusteringSequence),
lists:foreach(fun (Arg) ->
ok = control_action(reset, []),
- ok = control_action(cluster, Arg),
+ ok = control_action(force_cluster, Arg),
ok = control_action(start_app, []),
ok = control_action(stop_app, []),
ok
@@ -630,13 +630,13 @@ test_cluster_management() ->
ok = control_action(reset, []),
ok = control_action(start_app, []),
ok = control_action(stop_app, []),
- ok = control_action(cluster, ["invalid1@invalid",
- "invalid2@invalid"]),
+ ok = control_action(force_cluster, ["invalid1@invalid",
+ "invalid2@invalid"]),
%% join a non-existing cluster as a ram node
ok = control_action(reset, []),
- ok = control_action(cluster, ["invalid1@invalid",
- "invalid2@invalid"]),
+ ok = control_action(force_cluster, ["invalid1@invalid",
+ "invalid2@invalid"]),
SecondaryNode = rabbit_misc:makenode("hare"),
case net_adm:ping(SecondaryNode) of
@@ -661,18 +661,26 @@ test_cluster_management2(SecondaryNode) ->
%% join cluster as a ram node
ok = control_action(reset, []),
- ok = control_action(cluster, [SecondaryNodeS, "invalid1@invalid"]),
+ ok = control_action(force_cluster, [SecondaryNodeS, "invalid1@invalid"]),
ok = control_action(start_app, []),
ok = control_action(stop_app, []),
%% change cluster config while remaining in same cluster
- ok = control_action(cluster, ["invalid2@invalid", SecondaryNodeS]),
+ ok = control_action(force_cluster, ["invalid2@invalid", SecondaryNodeS]),
ok = control_action(start_app, []),
ok = control_action(stop_app, []),
%% join non-existing cluster as a ram node
- ok = control_action(cluster, ["invalid1@invalid",
- "invalid2@invalid"]),
+ ok = control_action(force_cluster, ["invalid1@invalid",
+ "invalid2@invalid"]),
+ ok = control_action(start_app, []),
+ ok = control_action(stop_app, []),
+
+ %% join empty cluster as a ram node
+ ok = control_action(cluster, []),
+ ok = control_action(start_app, []),
+ ok = control_action(stop_app, []),
+
%% turn ram node into disk node
ok = control_action(reset, []),
ok = control_action(cluster, [SecondaryNodeS, NodeS]),
@@ -680,8 +688,8 @@ test_cluster_management2(SecondaryNode) ->
ok = control_action(stop_app, []),
%% convert a disk node into a ram node
- ok = control_action(cluster, ["invalid1@invalid",
- "invalid2@invalid"]),
+ ok = control_action(force_cluster, ["invalid1@invalid",
+ "invalid2@invalid"]),
%% turn a disk node into a ram node
ok = control_action(reset, []),
@@ -1177,6 +1185,6 @@ handle_hook(HookName, Handler, Args) ->
A = atom_to_list(HookName) ++ "_" ++ atom_to_list(Handler) ++ "_fired",
put(list_to_atom(A), Args).
bad_handle_hook(_, _, _) ->
- bad:bad().
+ exit(bad_handle_hook_called).
extra_arg_hook(Hookname, Handler, Args, Extra1, Extra2) ->
handle_hook(Hookname, Handler, {Args, Extra1, Extra2}).
View Lorry Controller Status