Merge branch 'master' into rabbitmq-server-1246-master

Author: michaelklishin

docs/rabbitmqctl.8

@@ -1847,6 +1847,32 @@ For example:
 .sp
 .Dl rabbitmqctl encode --cipher blowfish_cfb64 --hash sha256 --iterations 10000 '<<"guest">>' mypassphrase
 .El
+.\" ------------------------------------
+.It Cm decode Ar value Ar passphrase
+.Bl -tag -width Ds
+.It Ar value Ar passphrase
+Value to decrypt (as produced by the encode command) and passphrase.
+.Pp
+For example:
+.sp
+.Dl rabbitmqctl decode '{encrypted, <<"...">>}' mypassphrase
+.El
+.\" ------------------------------------
+.It Cm list_hashes
+Lists hash functions supported by encoding commands.
+.Pp
+For example, this command instructs the RabbitMQ broker to list all hash
+functions supported by encoding commands:
+.sp
+.Dl rabbitmqctl list_hashes
+.\" ------------------------------------
+.It Cm list_ciphers
+Lists cipher suites supported by encoding commands.
+.Pp
+For example, this command instructs the RabbitMQ broker to list all
+cipher suites supported by encoding commands:
+.sp
+.Dl rabbitmqctl list_ciphers
 .El
 .\" ------------------------------------------------------------------
 .Sh PLUGIN COMMANDS

src/rabbit_amqqueue.erl

@@ -627,12 +627,12 @@ info_keys() -> rabbit_amqqueue_process:info_keys().
 map(Qs, F) -> rabbit_misc:filter_exit_map(F, Qs).
 
 info(Q = #amqqueue{ state = crashed }) -> info_down(Q, crashed);
-info(#amqqueue{ pid = QPid }) -> delegate:call(QPid, info).
+info(#amqqueue{ pid = QPid }) -> delegate:invoke(QPid, {gen_server2, call, [info, infinity]}).
 
 info(Q = #amqqueue{ state = crashed }, Items) ->
     info_down(Q, Items, crashed);
 info(#amqqueue{ pid = QPid }, Items) ->
-    case delegate:call(QPid, {info, Items}) of
+    case delegate:invoke(QPid, {gen_server2, call, [{info, Items}, infinity]}) of
         {ok, Res}      -> Res;
         {error, Error} -> throw(Error)
     end.
@@ -693,7 +693,8 @@ force_event_refresh(Ref) ->
 notify_policy_changed(#amqqueue{pid = QPid}) ->
     gen_server2:cast(QPid, policy_changed).
 
-consumers(#amqqueue{ pid = QPid }) -> delegate:call(QPid, consumers).
+consumers(#amqqueue{ pid = QPid }) ->
+    delegate:invoke(QPid, {gen_server2, call, [consumers, infinity]}).
 
 consumer_info_keys() -> ?CONSUMER_INFO_KEYS.
 
@@ -721,7 +722,7 @@ get_queue_consumer_info(Q, ConsumerInfoKeys) ->
                 AckRequired, Prefetch, Args]) ||
         {ChPid, CTag, AckRequired, Prefetch, Args, _} <- consumers(Q)].
 
-stat(#amqqueue{pid = QPid}) -> delegate:call(QPid, stat).
+stat(#amqqueue{pid = QPid}) -> delegate:invoke(QPid, {gen_server2, call, [stat, infinity]}).
 
 pid_of(#amqqueue{pid = Pid}) -> Pid.
 pid_of(VHost, QueueName) ->
@@ -739,7 +740,7 @@ delete_immediately(QPids) ->
     ok.
 
 delete(#amqqueue{ pid = QPid }, IfUnused, IfEmpty, ActingUser) ->
-    delegate:call(QPid, {delete, IfUnused, IfEmpty, ActingUser}).
+    delegate:invoke(QPid, {gen_server2, call, [{delete, IfUnused, IfEmpty, ActingUser}, infinity]}).
 
 delete_crashed(Q) ->
     delete_crashed(Q, ?INTERNAL_USER).
@@ -752,21 +753,24 @@ delete_crashed_internal(Q = #amqqueue{ name = QName }, ActingUser) ->
     BQ:delete_crashed(Q),
     ok = internal_delete(QName, ActingUser).
 
-purge(#amqqueue{ pid = QPid }) -> delegate:call(QPid, purge).
+purge(#amqqueue{ pid = QPid }) ->
+    delegate:invoke(QPid, {gen_server2, call, [purge, infinity]}).
 
-requeue(QPid, MsgIds, ChPid) -> delegate:call(QPid, {requeue, MsgIds, ChPid}).
+requeue(QPid, MsgIds, ChPid) ->
+    delegate:invoke(QPid, {gen_server2, call, [{requeue, MsgIds, ChPid}, infinity]}).
 
-ack(QPid, MsgIds, ChPid) -> delegate:cast(QPid, {ack, MsgIds, ChPid}).
+ack(QPid, MsgIds, ChPid) ->
+    delegate:invoke_no_result(QPid, {gen_server2, cast, [{ack, MsgIds, ChPid}]}).
 
 reject(QPid, Requeue, MsgIds, ChPid) ->
-    delegate:cast(QPid, {reject, Requeue, MsgIds, ChPid}).
+    delegate:invoke_no_result(QPid, {gen_server2, cast, [{reject, Requeue, MsgIds, ChPid}]}).
 
 notify_down_all(QPids, ChPid) ->
     notify_down_all(QPids, ChPid, ?CHANNEL_OPERATION_TIMEOUT).
 
 notify_down_all(QPids, ChPid, Timeout) ->
-    case rpc:call(node(), delegate, call,
-                  [QPids, {notify_down, ChPid}], Timeout) of
+    case rpc:call(node(), delegate, invoke,
+                  [QPids, {gen_server2, call, [{notify_down, ChPid}, infinity]}], Timeout) of
         {badrpc, timeout} -> {error, {channel_operation_timeout, Timeout}};
         {badrpc, Reason}  -> {error, Reason};
         {_, Bads} ->
@@ -782,35 +786,37 @@ notify_down_all(QPids, ChPid, Timeout) ->
     end.
 
 activate_limit_all(QPids, ChPid) ->
-    delegate:cast(QPids, {activate_limit, ChPid}).
+    delegate:invoke_no_result(QPids, {gen_server2, cast, [{activate_limit, ChPid}]}).
 
 credit(#amqqueue{pid = QPid}, ChPid, CTag, Credit, Drain) ->
-    delegate:cast(QPid, {credit, ChPid, CTag, Credit, Drain}).
+    delegate:invoke_no_result(QPid, {gen_server2, cast, [{credit, ChPid, CTag, Credit, Drain}]}).
 
 basic_get(#amqqueue{pid = QPid}, ChPid, NoAck, LimiterPid) ->
-    delegate:call(QPid, {basic_get, ChPid, NoAck, LimiterPid}).
+    delegate:invoke(QPid, {gen_server2, call, [{basic_get, ChPid, NoAck, LimiterPid}, infinity]}).
 
 basic_consume(#amqqueue{pid = QPid, name = QName}, NoAck, ChPid, LimiterPid,
               LimiterActive, ConsumerPrefetchCount, ConsumerTag,
               ExclusiveConsume, Args, OkMsg, ActingUser) ->
     ok = check_consume_arguments(QName, Args),
-    delegate:call(QPid, {basic_consume, NoAck, ChPid, LimiterPid, LimiterActive,
-                         ConsumerPrefetchCount, ConsumerTag, ExclusiveConsume,
-                         Args, OkMsg, ActingUser}).
+    delegate:invoke(QPid, {gen_server2, call,
+                           [{basic_consume, NoAck, ChPid, LimiterPid, LimiterActive,
+                             ConsumerPrefetchCount, ConsumerTag, ExclusiveConsume,
+                             Args, OkMsg, ActingUser}, infinity]}).
 
 basic_cancel(#amqqueue{pid = QPid}, ChPid, ConsumerTag, OkMsg, ActingUser) ->
-    delegate:call(QPid, {basic_cancel, ChPid, ConsumerTag, OkMsg, ActingUser}).
+    delegate:invoke(QPid, {gen_server2, call,
+                           [{basic_cancel, ChPid, ConsumerTag, OkMsg, ActingUser}, infinity]}).
 
 notify_decorators(#amqqueue{pid = QPid}) ->
-    delegate:cast(QPid, notify_decorators).
+    delegate:invoke_no_result(QPid, {gen_server2, cast, [notify_decorators]}).
 
 notify_sent(QPid, ChPid) ->
     rabbit_amqqueue_common:notify_sent(QPid, ChPid).
 
 notify_sent_queue_down(QPid) ->
     rabbit_amqqueue_common:notify_sent_queue_down(QPid).
 
-resume(QPid, ChPid) -> delegate:cast(QPid, {resume, ChPid}).
+resume(QPid, ChPid) -> delegate:invoke_no_result(QPid, {gen_server2, cast, [{resume, ChPid}]}).
 
 internal_delete1(QueueName, OnlyDurable) ->
     ok = mnesia:delete({rabbit_queue, QueueName}),
@@ -907,12 +913,17 @@ set_ram_duration_target(QPid, Duration) ->
 set_maximum_since_use(QPid, Age) ->
     gen_server2:cast(QPid, {set_maximum_since_use, Age}).
 
-update_mirroring(QPid) -> ok = delegate:cast(QPid, update_mirroring).
+update_mirroring(QPid) ->
+    ok = delegate:invoke_no_result(QPid, {gen_server2, cast, [update_mirroring]}).
 
-sync_mirrors(#amqqueue{pid = QPid}) -> delegate:call(QPid, sync_mirrors);
-sync_mirrors(QPid)                  -> delegate:call(QPid, sync_mirrors).
-cancel_sync_mirrors(#amqqueue{pid = QPid}) -> delegate:call(QPid, cancel_sync_mirrors);
-cancel_sync_mirrors(QPid)                  -> delegate:call(QPid, cancel_sync_mirrors).
+sync_mirrors(#amqqueue{pid = QPid}) ->
+    delegate:invoke(QPid, {gen_server2, call, [sync_mirrors, infinity]});
+sync_mirrors(QPid) ->
+    delegate:invoke(QPid, {gen_server2, call, [sync_mirrors, infinity]}).
+cancel_sync_mirrors(#amqqueue{pid = QPid}) ->
+    delegate:invoke(QPid, {gen_server2, call, [cancel_sync_mirrors, infinity]});
+cancel_sync_mirrors(QPid) ->
+    delegate:invoke(QPid, {gen_server2, call, [cancel_sync_mirrors, infinity]}).
 
 is_mirrored(Q) ->
     rabbit_mirror_queue_misc:is_mirrored(Q).
@@ -1031,8 +1042,8 @@ deliver(Qs, Delivery = #delivery{flow = Flow}) ->
     %% done with it.
     MMsg = {deliver, Delivery, false},
     SMsg = {deliver, Delivery, true},
-    delegate:cast(MPids, MMsg),
-    delegate:cast(SPids, SMsg),
+    delegate:invoke_no_result(MPids, {gen_server2, cast, [MMsg]}),
+    delegate:invoke_no_result(SPids, {gen_server2, cast, [SMsg]}),
     QPids.
 
 qpids([]) -> {[], []}; %% optimisation

src/rabbit_control_pbe.erl

@@ -16,61 +16,74 @@
 
 -module(rabbit_control_pbe).
 
--export([encode/7]).
+-export([decode/4, encode/4, list_ciphers/0, list_hashes/0]).
 
 % for testing purposes
 -export([evaluate_input_as_term/1]).
 
-encode(ListCiphers, _ListHashes, _Decode, _Cipher, _Hash, _Iterations, _Args) when ListCiphers ->
-    {ok, io_lib:format("~p", [rabbit_pbe:supported_ciphers()])};
+list_ciphers() ->
+    {ok, io_lib:format("~p", [rabbit_pbe:supported_ciphers()])}.
 
-encode(_ListCiphers, ListHashes, _Decode, _Cipher, _Hash, _Iterations, _Args) when ListHashes ->
-    {ok, io_lib:format("~p", [rabbit_pbe:supported_hashes()])};
+list_hashes() ->
+    {ok, io_lib:format("~p", [rabbit_pbe:supported_hashes()])}.
 
-encode(_ListCiphers, _ListHashes, Decode, Cipher, Hash, Iterations, Args) ->
-    CipherExists = lists:member(Cipher, rabbit_pbe:supported_ciphers()),
-    HashExists = lists:member(Hash, rabbit_pbe:supported_hashes()),
-    encode_encrypt_decrypt(CipherExists, HashExists, Decode, Cipher, Hash, Iterations, Args).
-
-encode_encrypt_decrypt(CipherExists, _HashExists, _Decode, _Cipher, _Hash, _Iterations, _Args) when CipherExists =:= false ->
-    {error, io_lib:format("The requested cipher is not supported", [])};
-
-encode_encrypt_decrypt(_CipherExists, HashExists, _Decode, _Cipher, _Hash, _Iterations, _Args) when HashExists =:= false ->
-    {error, io_lib:format("The requested hash is not supported", [])};
-
-encode_encrypt_decrypt(_CipherExists, _HashExists, _Decode, _Cipher, _Hash, Iterations, _Args) when Iterations =< 0 ->
+validate(_Cipher, _Hash, Iterations, _Args) when Iterations =< 0 ->
     {error, io_lib:format("The requested number of iterations is incorrect", [])};
+validate(_Cipher, _Hash, _Iterations, Args) when length(Args) < 2 ->
+    {error, io_lib:format("Please provide a value to encode/decode and a passphrase", [])};
+validate(_Cipher, _Hash, _Iterations, Args) when length(Args) > 2 ->
+    {error, io_lib:format("Too many arguments. Please provide a value to encode/decode and a passphrase", [])};
+validate(Cipher, Hash, _Iterations, _Args) ->
+    case lists:member(Cipher, rabbit_pbe:supported_ciphers()) of
+        false ->
+            {error, io_lib:format("The requested cipher is not supported", [])};
+        true ->
+            case lists:member(Hash, rabbit_pbe:supported_hashes()) of
+                false ->
+                    {error, io_lib:format("The requested hash is not supported", [])};
+                true -> ok
+            end
+    end.
 
-encode_encrypt_decrypt(_CipherExists, _HashExists, Decode, Cipher, Hash, Iterations, Args) when length(Args) == 2, Decode =:= false ->
-    [Value, PassPhrase] = Args,
-    try begin
-            TermValue = evaluate_input_as_term(Value),
-            Result = rabbit_pbe:encrypt_term(Cipher, Hash, Iterations, list_to_binary(PassPhrase), TermValue),
-            {ok, io_lib:format("~p", [{encrypted, Result}])}
-        end
-    catch
-        _:Msg -> {error, io_lib:format("Error during cipher operation: ~p", [Msg])}
-    end;
-
-encode_encrypt_decrypt(_CipherExists, _HashExists, Decode, Cipher, Hash, Iterations, Args) when length(Args) == 2, Decode ->
-    [Value, PassPhrase] = Args,
-    try begin
-            TermValue = evaluate_input_as_term(Value),
-            TermToDecrypt = case TermValue of
-                {encrypted, EncryptedTerm} ->
-                    EncryptedTerm;
-                _ ->
-                    TermValue
-            end,
-            Result = rabbit_pbe:decrypt_term(Cipher, Hash, Iterations, list_to_binary(PassPhrase), TermToDecrypt),
-            {ok, io_lib:format("~p", [Result])}
-        end
-    catch
-        _:Msg -> {error, io_lib:format("Error during cipher operation: ~p", [Msg])}
-    end;
+encode(Cipher, Hash, Iterations, Args) ->
+    case validate(Cipher, Hash, Iterations, Args) of
+        {error, Err} -> {error, Err};
+        ok ->
+            [Value, PassPhrase] = Args,
+            try begin
+                    TermValue = evaluate_input_as_term(Value),
+                    Result = rabbit_pbe:encrypt_term(Cipher, Hash, Iterations,
+                                                     list_to_binary(PassPhrase),
+                                                     TermValue),
+                    {ok, io_lib:format("~p", [{encrypted, Result}])}
+                end
+            catch
+                _:Msg -> {error, io_lib:format("Error during cipher operation: ~p", [Msg])}
+            end
+    end.
 
-encode_encrypt_decrypt(_CipherExists, _HashExists, _Decode, _Cipher, _Hash, _Iterations, _Args) ->
-    {error, io_lib:format("Please provide a value to encode/decode and a passphrase", [])}.
+decode(Cipher, Hash, Iterations, Args) ->
+    case validate(Cipher, Hash, Iterations, Args) of
+        {error, Err} -> {error, Err};
+        ok ->
+            [Value, PassPhrase] = Args,
+            try begin
+                    TermValue = evaluate_input_as_term(Value),
+                    TermToDecrypt = case TermValue of
+                        {encrypted, EncryptedTerm} ->
+                            EncryptedTerm;
+                        _ ->
+                            TermValue
+                    end,
+                    Result = rabbit_pbe:decrypt_term(Cipher, Hash, Iterations,
+                                                     list_to_binary(PassPhrase),
+                                                     TermToDecrypt),
+                    {ok, io_lib:format("~p", [Result])}
+                end
+            catch
+                _:Msg -> {error, io_lib:format("Error during cipher operation: ~p", [Msg])}
+            end
+    end.
 
 evaluate_input_as_term(Input) ->
     {ok,Tokens,_EndLine} = erl_scan:string(Input ++ "."),

src/rabbit_exchange_parameters.erl

@@ -23,8 +23,6 @@
 -export([register/0]).
 -export([validate/5, notify/5, notify_clear/4]).
 
--import(rabbit_misc, [pget/2]).
-
 -rabbit_boot_step({?MODULE,
                    [{description, "exchange parameters"},
                     {mfa, {rabbit_exchange_parameters, register, []}},
@@ -36,7 +34,8 @@ register() ->
                              ?EXCHANGE_DELETE_IN_PROGRESS_COMPONENT, ?MODULE),
     %% ensure there are no leftovers from before node restart/crash
     rabbit_runtime_parameters:clear_component(
-      ?EXCHANGE_DELETE_IN_PROGRESS_COMPONENT),
+      ?EXCHANGE_DELETE_IN_PROGRESS_COMPONENT,
+      ?INTERNAL_USER),
     ok.
 
 validate(_VHost, ?EXCHANGE_DELETE_IN_PROGRESS_COMPONENT, _Name, _Term, _User) ->

src/rabbit_looking_glass.erl

@@ -16,6 +16,8 @@
 
 -module(rabbit_looking_glass).
 
+-ignore_xref([{lg, trace, 4}]).
+
 -export([boot/0]).
 -export([connections/0]).
 

src/rabbit_runtime_parameters.erl

@@ -53,7 +53,7 @@
 
 -export([parse_set/5, set/5, set_any/5, clear/4, clear_any/4, list/0, list/1,
          list_component/1, list/2, list_formatted/1, list_formatted/3,
-         lookup/3, value/3, value/4, info_keys/0, clear_component/1]).
+         lookup/3, value/3, value/4, info_keys/0, clear_component/2]).
 
 -export([parse_set_global/3, set_global/3, value_global/1, value_global/2,
          list_global/0, list_global_formatted/0, list_global_formatted/2,
@@ -95,7 +95,7 @@
 
 %%---------------------------------------------------------------------------
 
--import(rabbit_misc, [pget/2, pset/3]).
+-import(rabbit_misc, [pget/2]).
 
 -define(TABLE, rabbit_runtime_parameters).
 
@@ -228,14 +228,15 @@ clear_global(Key, ActingUser) ->
             end
     end.
 
-clear_component(Component) ->
-    case rabbit_runtime_parameters:list_component(Component) of
+clear_component(Component, ActingUser) ->
+    case list_component(Component) of
         [] ->
             ok;
         Xs ->
-            [rabbit_runtime_parameters:clear(pget(vhost, X),
-                                             pget(component, X),
-                                             pget(name, X))|| X <- Xs],
+            [clear(pget(vhost, X),
+                   pget(component, X),
+                   pget(name, X),
+                   ActingUser) || X <- Xs],
             ok
     end.
 

test/cluster_SUITE.erl

@@ -123,11 +123,6 @@ delegates_async1(_Config, SecondaryNode) ->
     ok = delegate:invoke_no_result(spawn(SecondaryNode, Responder), Sender),
     await_response(2),
 
-    LocalPids = spawn_responders(node(), Responder, 10),
-    RemotePids = spawn_responders(SecondaryNode, Responder, 10),
-    ok = delegate:invoke_no_result(LocalPids ++ RemotePids, Sender),
-    await_response(20),
-
     passed.
 
 delegates_sync(Config) ->