ref 2

Author: ansd

deps/rabbit/src/rabbit_volatile_queue.erl

@@ -20,11 +20,12 @@
          is/1,
          key_from_name/1,
          pid_from_name/2,
-         local_cast/2,
-         local_call/2,
          exists/1,
-         ff_enabled/0]).
+         ff_enabled/0,
+         local_cast/2,
+         local_call/2]).
 
+%% rabbit_queue_type callbacks
 -export([declare/2,
          supports_stateful_delivery/0,
          deliver/3,
@@ -79,28 +80,25 @@
 -spec new(rabbit_amqqueue:name()) ->
     amqqueue:amqqueue() | error.
 new(#resource{virtual_host = Vhost,
-              name = <<"amq.rabbitmq.reply-to">>} = QName) ->
-    new0(QName, self(), Vhost);
+              name = <<"amq.rabbitmq.reply-to">>} = Name) ->
+    new0(Name, self(), Vhost);
 new(#resource{virtual_host = Vhost,
-              name = QNameBin} = QName) ->
-    case pid_from_name(QNameBin, nodes_with_hashes()) of
+              name = NameBin} = Name) ->
+    case pid_from_name(NameBin, nodes_with_hashes()) of
         {ok, Pid} when is_pid(Pid) ->
-            new0(QName, Pid, Vhost);
+            new0(Name, Pid, Vhost);
         _ ->
             error
     end.
 
-new0(QName, Pid, Vhost) ->
-    amqqueue:new(QName, Pid, false, true, none, [], Vhost, #{}, ?MODULE).
-
-nodes_with_hashes() ->
-    #{erlang:phash2(Node) => Node || Node <- rabbit_nodes:list_members()}.
+new0(Name, Pid, Vhost) ->
+    amqqueue:new(Name, Pid, false, true, none, [], Vhost, #{}, ?MODULE).
 
 -spec is(rabbit_misc:resource_name()) ->
     boolean().
 is(<<?PREFIX, _/binary>>) ->
     true;
-is(_) ->
+is(Name) when is_binary(Name) ->
     false.
 
 init(Q) ->
@@ -121,15 +119,15 @@ consume(_Q, Spec, State) ->
                       credit = Credit}}.
 
 declare(Q, _Node) ->
-    #resource{name = QNameBin} = QName = amqqueue:get_name(Q),
-    case QNameBin of
+    #resource{name = NameBin} = Name = amqqueue:get_name(Q),
+    case NameBin of
         <<"amq.rabbitmq.reply-to">> ->
             {existing, Q};
         _ ->
-            case exists(QName) of
+            case exists(Name) of
                 true ->
                     {existing, Q};
-                _ ->
+                false ->
                     {absent, Q, stopped}
             end
     end.
@@ -142,7 +140,6 @@ exists(#resource{kind = queue,
             case ff_enabled() of
                 true ->
                     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
@@ -187,7 +184,6 @@ deliver0(Q, Msg) ->
     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, local_cast, [Request]},
             delegate:invoke_no_result(QPid, MFA);
         false ->
@@ -253,20 +249,23 @@ credit(_QName, CTag, DeliveryCountRcv, LinkCreditRcv, Drain,
                               end,
     {State#?STATE{delivery_count = DeliveryCount,
                   credit = Credit},
-     [{credit_reply, CTag, DeliveryCount, Credit, 0, Drain}]}.
+     [{credit_reply, CTag, DeliveryCount, Credit, _Available = 0, Drain}]}.
 
 close(#?STATE{}) ->
     ok.
 
-update(_, State) ->
+update(_, #?STATE{} = State) ->
     State.
 
-cancel(_, _, State) ->
+cancel(_, _, #?STATE{} = State) ->
     {ok, State}.
 
 is_enabled() ->
     true.
 
+ff_enabled() ->
+    rabbit_feature_flags:is_enabled('rabbitmq_4.2.0').
+
 is_compatible(_, _, _) ->
     true.
 
@@ -322,18 +321,18 @@ delete(_, _, _, _) ->
 recover(_, _) ->
     {[], []}.
 
-settle(_, _, _, _, State) ->
+settle(_, _, _, _, #?STATE{} = State) ->
     {State, []}.
 
-credit_v1(_, _, _, _, State) ->
+credit_v1(_, _, _, _, #?STATE{} = State) ->
     {State, []}.
 
 dequeue(_, _, _, _, #?STATE{name = Name}) ->
     {protocol_error, not_implemented,
      "basic.get not supported by volatile ~ts",
      [rabbit_misc:rs(Name)]}.
 
-state_info(_) ->
+state_info(#?STATE{}) ->
     #{}.
 
 info(_, _) ->
@@ -342,9 +341,6 @@ info(_, _) ->
 policy_apply_to_name() ->
     <<>>.
 
-ff_enabled() ->
-    rabbit_feature_flags:is_enabled('rabbitmq_4.2.0').
-
 -spec new_name() ->
     rabbit_misc:resource_name().
 new_name() ->
@@ -408,3 +404,6 @@ key_from_name(<<?PREFIX, Suffix/binary>>) ->
     end;
 key_from_name(_) ->
     error.
+
+nodes_with_hashes() ->
+    #{erlang:phash2(Node) => Node || Node <- rabbit_nodes:list_members()}.

deps/rabbit_common/src/delegate.erl

@@ -18,9 +18,8 @@
 %% consistent route, to prevent them being reordered. In fact all
 %% AMQP-ish things (such as queue declaration results and basic.get)
 %% must take the same route as well, to ensure that clients see causal
-%% ordering correctly. Therefore we have a rather generic mechanism
-%% here rather than just a message-reflector. That's also why we pick
-%% the delegate process to use based on a hash of the source pid.
+%% ordering correctly. Therefore we can't use erpc. That's also why we
+%% pick the delegate process to use based on a hash of the source pid.
 %%
 %% When a function is invoked using delegate:invoke/2,
 %% or delegate:invoke_no_result/2 on a group of pids, the pids are first split