Merge branch 'master' into rabbitmq-server-567

3 files changed, 145 insertions, 9 deletions

diff --git a/src/rabbit_control_main.erl b/src/rabbit_control_main.erl
index babaf4fd4e..d96c1dd476 100644
--- a/src/rabbit_control_main.erl
+++ b/src/rabbit_control_main.erl
@@ -70,6 +70,10 @@
          {clear_parameter, [?VHOST_DEF]},
          {list_parameters, [?VHOST_DEF]},
 
+         set_global_parameter,
+         clear_global_parameter,
+         list_global_parameters,
+
          {set_policy, [?VHOST_DEF, ?PRIORITY_DEF, ?APPLY_TO_DEF]},
          {clear_policy, [?VHOST_DEF]},
          {set_operator_policy, [?VHOST_DEF, ?PRIORITY_DEF, ?APPLY_TO_DEF]},
@@ -126,7 +130,7 @@
 -define(COMMANDS_WITH_TIMEOUT,
         [list_user_permissions, list_policies, list_queues, list_exchanges,
          list_bindings, list_connections, list_channels, list_consumers,
-         list_vhosts, list_parameters,
+         list_vhosts, list_parameters, list_global_parameters,
          purge_queue,
          {node_health_check, 70000}]).
 
@@ -282,7 +286,7 @@ action(stop, Node, Args, _Opts, Inform) ->
     Res;
 
 action(stop_app, Node, [], _Opts, Inform) ->
-    Inform("Stopping node ~p", [Node]),
+    Inform("Stopping rabbit application on node ~p", [Node]),
     call(Node, {rabbit, stop, []});
 
 action(start_app, Node, [], _Opts, Inform) ->
@@ -529,6 +533,20 @@ action(clear_parameter, Node, [Component, Key], Opts, Inform) ->
                                                       list_to_binary(Component),
                                                       list_to_binary(Key)]);
 
+action(set_global_parameter, Node, [Key, Value], _Opts, Inform) ->
+    Inform("Setting global runtime parameter ~p to ~p", [Key, Value]),
+    rpc_call(
+        Node, rabbit_runtime_parameters, parse_set_global,
+        [rabbit_data_coercion:to_atom(Key), rabbit_data_coercion:to_binary(Value)]
+    );
+
+action(clear_global_parameter, Node, [Key], _Opts, Inform) ->
+    Inform("Clearing global runtime parameter ~p", [Key]),
+    rpc_call(
+        Node, rabbit_runtime_parameters, clear_global,
+        [rabbit_data_coercion:to_atom(Key)]
+    );
+
 action(set_policy, Node, [Key, Pattern, Defn], Opts, Inform) ->
     Msg = "Setting policy ~p for pattern ~p to ~p with priority ~p",
     VHostArg = list_to_binary(proplists:get_value(?VHOST_OPT, Opts)),
@@ -659,6 +677,12 @@ action(list_parameters, Node, [], Opts, Inform, Timeout) ->
                  rabbit_runtime_parameters:info_keys(),
                  [{timeout, Timeout}]);
 
+action(list_global_parameters, Node, [], _Opts, Inform, Timeout) ->
+    Inform("Listing global runtime parameters", []),
+    call_emitter(Node, {rabbit_runtime_parameters, list_global_formatted, []},
+         rabbit_runtime_parameters:global_info_keys(),
+         [{timeout, Timeout}]);
+
 action(list_policies, Node, [], Opts, Inform, Timeout) ->
     VHostArg = list_to_binary(proplists:get_value(?VHOST_OPT, Opts)),
     Inform("Listing policies", []),
diff --git a/src/rabbit_runtime_parameters.erl b/src/rabbit_runtime_parameters.erl
index 072a48be3d..94018a5b54 100644
--- a/src/rabbit_runtime_parameters.erl
+++ b/src/rabbit_runtime_parameters.erl
@@ -40,6 +40,8 @@
 %% Parameters are stored in Mnesia and can be global. Their changes
 %% are broadcasted over rabbit_event.
 %%
+%% Global parameters keys are atoms and values are JSON documents.
+%%
 %% See also:
 %%
 %%  * rabbit_policies
@@ -53,7 +55,9 @@
          list_component/1, list/2, list_formatted/1, list_formatted/3,
          lookup/3, value/3, value/4, info_keys/0, clear_component/1]).
 
--export([set_global/2, value_global/1, value_global/2]).
+-export([parse_set_global/2, set_global/2, value_global/1, value_global/2,
+         list_global/0, list_global_formatted/0, list_global_formatted/2,
+         lookup_global/1, global_info_keys/0, clear_global/1]).
 
 %%----------------------------------------------------------------------------
 
@@ -109,10 +113,19 @@ set(_, <<"policy">>, _, _, _) ->
 set(VHost, Component, Name, Term, User) ->
     set_any(VHost, Component, Name, Term, User).
 
-set_global(Name, Term) ->
-    mnesia_update(Name, Term),
-    event_notify(parameter_set, none, global, [{name,  Name},
-                                               {value, Term}]),
+parse_set_global(Name, String) ->
+    Definition = rabbit_data_coercion:to_binary(String),
+    case rabbit_json:try_decode(Definition) of
+        {ok, Term} when is_map(Term) -> set_global(Name, maps:to_list(Term));
+        {ok, Term} -> set_global(Name, Term);
+        error      -> {error_string, "JSON decoding error"}
+    end.
+
+set_global(Name, Term)  ->
+    NameAsAtom = rabbit_data_coercion:to_atom(Name),
+    mnesia_update(NameAsAtom, Term),
+    event_notify(parameter_set, none, global, [{name,  NameAsAtom},
+        {value, Term}]),
     ok.
 
 format_error(L) ->
@@ -168,6 +181,26 @@ clear(_, <<"policy">> , _) ->
 clear(VHost, Component, Name) ->
     clear_any(VHost, Component, Name).
 
+clear_global(Key) ->
+    KeyAsAtom = rabbit_data_coercion:to_atom(Key),
+    Notify = fun() ->
+                    event_notify(parameter_set, none, global, [{name,  KeyAsAtom}]),
+                    ok
+             end,
+    case value_global(KeyAsAtom) of
+        not_found ->
+            {error_string, "Parameter does not exist"};
+        _         ->
+            F = fun () ->
+                ok = mnesia:delete(?TABLE, KeyAsAtom, write)
+                end,
+            ok = rabbit_misc:execute_mnesia_transaction(F),
+            case mnesia:is_transaction() of
+                true  -> Notify;
+                false -> Notify()
+            end
+    end.
+
 clear_component(Component) ->
     case rabbit_runtime_parameters:list_component(Component) of
         [] ->
@@ -235,6 +268,15 @@ list(VHost, Component) ->
                        Comp =/= <<"policy">> orelse Component =:= <<"policy">>]
       end).
 
+list_global() ->
+    %% list only atom keys
+    mnesia:async_dirty(
+        fun () ->
+            Match = #runtime_parameters{key = '_', _ = '_'},
+            [p(P) || P <- mnesia:match_object(?TABLE, Match, read),
+                is_atom(P#runtime_parameters.key)]
+        end).
+
 list_formatted(VHost) ->
     [pset(value, rabbit_json:encode(pget(value, P)), P) || P <- list(VHost)].
 
@@ -243,17 +285,34 @@ list_formatted(VHost, Ref, AggregatorPid) ->
       AggregatorPid, Ref,
       fun(P) -> pset(value, rabbit_json:encode(pget(value, P)), P) end, list(VHost)).
 
+list_global_formatted() ->
+    [pset(value, rabbit_json:encode(pget(value, P)), P) || P <- list_global()].
+
+list_global_formatted(Ref, AggregatorPid) ->
+    rabbit_control_misc:emitting_map(
+        AggregatorPid, Ref,
+        fun(P) -> pset(value, rabbit_json:encode(pget(value, P)), P) end, list_global()).
+
 lookup(VHost, Component, Name) ->
     case lookup0({VHost, Component, Name}, rabbit_misc:const(not_found)) of
         not_found -> not_found;
         Params    -> p(Params)
     end.
 
+lookup_global(Name)  ->
+    case lookup0(Name, rabbit_misc:const(not_found)) of
+        not_found -> not_found;
+        Params    -> p(Params)
+    end.
+
 value(VHost, Comp, Name)      -> value0({VHost, Comp, Name}).
 value(VHost, Comp, Name, Def) -> value0({VHost, Comp, Name}, Def).
 
-value_global(Key)          -> value0(Key).
-value_global(Key, Default) -> value0(Key, Default).
+value_global(Key) ->
+    value0(Key).
+
+value_global(Key, Default) ->
+    value0(Key, Default).
 
 value0(Key) ->
     case lookup0(Key, rabbit_misc:const(not_found)) of
@@ -290,10 +349,16 @@ p(#runtime_parameters{key = {VHost, Component, Name}, value = Value}) ->
     [{vhost,     VHost},
      {component, Component},
      {name,      Name},
+     {value,     Value}];
+
+p(#runtime_parameters{key = Key, value = Value}) when is_atom(Key) ->
+    [{name,      Key},
      {value,     Value}].
 
 info_keys() -> [component, name, value].
 
+global_info_keys() -> [name, value].
+
 %%---------------------------------------------------------------------------
 
 lookup_component(Component) ->
diff --git a/src/worker_pool_worker.erl b/src/worker_pool_worker.erl
index bd07f0d782..c515abe07f 100644
--- a/src/worker_pool_worker.erl
+++ b/src/worker_pool_worker.erl
@@ -27,6 +27,7 @@
          run/1]).
 
 -export([set_maximum_since_use/2]).
+-export([set_timeout/2, set_timeout/3, clear_timeout/1]).
 
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2,
          terminate/2, code_change/3, prioritise_cast/3]).
@@ -136,6 +137,11 @@ handle_info({'DOWN', MRef, process, CPid, _Reason}, {from, CPid, MRef}) ->
 handle_info({'DOWN', _MRef, process, _Pid, _Reason}, State) ->
     {noreply, State, hibernate};
 
+handle_info({timeout, Key, Fun}, State) ->
+    clear_timeout(Key),
+    Fun(),
+    {noreply, State, hibernate};
+
 handle_info(Msg, State) ->
     {stop, {unexpected_info, Msg}, State}.
 
@@ -144,3 +150,44 @@ code_change(_OldVsn, State, _Extra) ->
 
 terminate(_Reason, State) ->
     State.
+
+-spec set_timeout(integer(), fun(() -> any())) -> reference().
+set_timeout(Time, Fun) ->
+    Key = make_ref(),
+    set_timeout(Key, Time, Fun).
+
+-spec set_timeout(Key, integer(), fun(() -> any())) -> Key when Key :: any().
+set_timeout(Key, Time, Fun) ->
+    Timeouts = get_timeouts(),
+    set_timeout(Key, Time, Fun, Timeouts).
+
+-spec clear_timeout(any()) -> ok.
+clear_timeout(Key) ->
+    NewTimeouts = cancel_timeout(Key, get_timeouts()),
+    put(timeouts, NewTimeouts),
+    ok.
+
+get_timeouts() ->
+    case get(timeouts) of
+        undefined -> dict:new();
+        Dict      -> Dict
+    end.
+
+set_timeout(Key, Time, Fun, Timeouts) ->
+    cancel_timeout(Key, Timeouts),
+    {ok, TRef} = timer:send_after(Time, {timeout, Key, Fun}),
+    NewTimeouts = dict:store(Key, TRef, Timeouts),
+    put(timeouts, NewTimeouts),
+    {ok, Key}.
+
+cancel_timeout(Key, Timeouts) ->
+    case dict:find(Key, Timeouts) of
+        {ok, TRef} ->
+            timer:cancel(TRef),
+            receive {timeout, Key, _} -> ok
+            after 0 -> ok
+            end,
+            dict:erase(Key, Timeouts);
+        error ->
+            Timeouts
+    end.