Use delegate for queue existence check

Author: ansd

deps/rabbit/src/rabbit_volatile_queue.erl

@@ -20,7 +20,8 @@
          generate_name/1,
          key_from_name/1,
          pid_from_name/2,
-         deliver_local/3,
+         local_cast/2,
+         local_call/2,
          exists/1,
          ff_enabled/0]).
 
@@ -83,9 +84,9 @@ new(#resource{virtual_host = Vhost,
 new(#resource{virtual_host = Vhost,
               name = QNameBin} = QName) ->
     case pid_from_name(QNameBin, nodes_with_hashes()) of
-        {ok, Pid} ->
+        {ok, Pid} when is_pid(Pid) ->
             new0(QName, Pid, Vhost);
-        error ->
+        _ ->
             error
     end.
 
@@ -137,31 +138,29 @@ declare(Q, _Node) ->
 exists(#resource{kind = queue,
                  name = QNameBin} = QName) ->
     case pid_from_name(QNameBin, nodes_with_hashes()) of
-        {ok, Pid} ->
+        {ok, Pid} when is_pid(Pid) ->
             case ff_enabled() of
                 true ->
-                    Msg = {has_state, QName, ?MODULE},
-                    try gen_server:call(Pid, Msg)
-                    catch exit:_ ->
-                              false
+                    Request = {has_state, QName, ?MODULE},
+                    %% We use delegate instead of erpc to guarantee that order is preserved.
+                    MFA = {?MODULE, local_call, [Request]},
+                    try delegate:invoke(Pid, MFA)
+                    catch _:_ -> false
                     end;
                 false ->
                     case key_from_name(QNameBin) of
                         {ok, Key} ->
                             Msg = {declare_fast_reply_to, Key},
                             try gen_server:call(Pid, Msg, infinity) of
-                                exists ->
-                                    true;
-                                not_found ->
-                                    false
-                            catch exit:_ ->
-                                      false
+                                exists -> true;
+                                _ -> false
+                            catch exit:_ -> false
                             end;
                         error ->
                             false
                     end
             end;
-        error ->
+        _ ->
             false
     end.
 
@@ -187,8 +186,9 @@ deliver0(Q, Msg) ->
     QPid = amqqueue:get_pid(Q),
     case ff_enabled() of
         true ->
+            Request = {queue_event, QName, {deliver, Msg}},
             %% We use delegate instead of erpc to guarantee that order is preserved.
-            MFA = {?MODULE, deliver_local, [QName, Msg]},
+            MFA = {?MODULE, local_cast, [Request]},
             delegate:invoke_no_result(QPid, MFA);
         false ->
             case key_from_name(QName#resource.name) of
@@ -200,18 +200,26 @@ deliver0(Q, Msg) ->
             end
     end.
 
--spec deliver_local(pid(), rabbit_amqqueue:name(), mc:state()) -> ok.
-deliver_local(Pid, QName, Msg) ->
-    %% We ensure clients can't use this mechanism to send a message
-    %% to an arbitrary process and kill it.
-    case rabbit_amqp_session:is_local(Pid) orelse
-         pg_local:in_group(rabbit_channels, Pid) of
-        true ->
-            gen_server:cast(Pid, {queue_event, QName, {deliver, Msg}});
-        false ->
-            ok
+-spec local_cast(pid(), term()) -> ok.
+local_cast(Pid, Request) ->
+    %% Ensure clients can't send a message to an arbitrary process and kill it.
+    case is_local(Pid) of
+        true -> gen_server:cast(Pid, Request);
+        false -> ok
     end.
 
+-spec local_call(pid(), term()) -> term().
+local_call(Pid, Request) ->
+    %% Ensure clients can't send a message to an arbitrary process and kill it.
+    case is_local(Pid) of
+        true -> gen_server:call(Pid, Request);
+        false -> exit({unknown_pid, Pid})
+    end.
+
+is_local(Pid) ->
+    rabbit_amqp_session:is_local(Pid) orelse
+    pg_local:in_group(rabbit_channels, Pid).
+
 handle_event(QName, {deliver, Msg}, #?STATE{name = QName,
                                             ctag = Ctag,
                                             credit = undefined} = State) ->
@@ -383,9 +391,7 @@ pid_from_name(<<?PREFIX, Bin/binary>>, CandidateNodes) ->
                 PidParts = maps:update(node, Candidate, PidParts0),
                 {ok, rabbit_pid_codec:recompose(PidParts)}
         end
-    catch
-        error:_ ->
-            error
+    catch error:_ -> error
     end;
 pid_from_name(_, _) ->
     error.