ref

Author: ansd

deps/rabbit/src/rabbit_amqp_session.erl

@@ -90,9 +90,9 @@
 -define(MAX_PERMISSION_CACHE_SIZE, 12).
 -define(HIBERNATE_AFTER, 6_000).
 -define(CREDIT_REPLY_TIMEOUT, 30_000).
--define(CAP_RABBITMQ_VOLATILE_QUEUE, <<"rabbitmq:volatile-queue">>).
 %% Capability defined in amqp-bindmap-jms-v1.0-wd10 [5.2] and sent by Qpid JMS client.
 -define(CAP_TEMPORARY_QUEUE, <<"temporary-queue">>).
+-define(CAP_VOLATILE_QUEUE, <<"rabbitmq:volatile-queue">>).
 
 -export([start_link/9,
          process_frame/2,
@@ -2756,14 +2756,14 @@ ensure_source(#'v1_0.source'{
               ConnPid, PermCache0, TopicPermCache) ->
     FFEnabled = rabbit_volatile_queue:ff_enabled(),
     case maps:from_keys(Caps, true) of
-        #{{symbol, ?CAP_RABBITMQ_VOLATILE_QUEUE} := true}
+        #{{symbol, ?CAP_VOLATILE_QUEUE} := true}
           when (Durable =:= undefined orelse Durable =:= ?V_1_0_TERMINUS_DURABILITY_NONE) andalso
                ExpiryPolicy =:= ?V_1_0_TERMINUS_EXPIRY_POLICY_LINK_DETACH andalso
                (Timeout =:= undefined orelse Timeout =:= {uint, 0}) andalso
                SndSettled andalso
                FFEnabled ->
             %% create volatile queue
-            QNameBin = rabbit_volatile_queue:generate_name(self()),
+            QNameBin = rabbit_volatile_queue:new_name(),
             Source = #'v1_0.source'{
                         address = {utf8, queue_address(QNameBin)},
                         durable = Durable,
@@ -2772,7 +2772,7 @@ ensure_source(#'v1_0.source'{
                         dynamic = true,
                         dynamic_node_properties = dynamic_node_properties(),
                         distribution_mode = ?V_1_0_STD_DIST_MODE_MOVE,
-                        capabilities = rabbit_amqp_util:capabilities([?CAP_RABBITMQ_VOLATILE_QUEUE])
+                        capabilities = rabbit_amqp_util:capabilities([?CAP_VOLATILE_QUEUE])
                        },
             QName = rabbit_misc:queue_resource(Vhost, QNameBin),
             {ok, QName, Source, PermCache0, TopicPermCache};
@@ -3521,30 +3521,15 @@ error_if_absent(Kind, Vhost, Name) when is_list(Name) ->
 error_if_absent(Kind, Vhost, Name) when is_binary(Name) ->
     error_if_absent(rabbit_misc:r(Vhost, Kind, Name)).
 
-error_if_absent(#resource{kind = exchange} = XName) ->
-    case rabbit_exchange:exists(XName) of
-        true ->
-            ok;
-        false ->
-            link_error_not_found(XName)
+error_if_absent(#resource{kind = exchange} = Name) ->
+    case rabbit_exchange:exists(Name) of
+        true -> ok;
+        false -> link_error_not_found(Name)
     end;
-error_if_absent(#resource{kind = queue,
-                          name = QNameBin} = QName) ->
-    case rabbit_volatile_queue:is(QNameBin) of
-        true ->
-            case rabbit_volatile_queue:exists(QName) of
-                true ->
-                    ok;
-                false ->
-                    link_error_not_found(QName)
-            end;
-        false ->
-            case rabbit_amqqueue:exists(QName) of
-                true ->
-                    ok;
-                false ->
-                    link_error_not_found(QName)
-            end
+error_if_absent(#resource{kind = queue} = Name) ->
+    case rabbit_amqqueue:exists(Name) of
+        true -> ok;
+        false -> link_error_not_found(Name)
     end.
 
 generate_queue_name_v1() ->

deps/rabbit/src/rabbit_channel.erl

@@ -1307,7 +1307,7 @@ handle_method(#'basic.consume'{queue        = <<"amq.rabbitmq.reply-to">>,
                                Other ->
                                    Other
                            end,
-                    QNameBin = rabbit_volatile_queue:generate_name(self()),
+                    QNameBin = rabbit_volatile_queue:new_name(),
                     QName = rabbit_misc:queue_resource(Vhost, QNameBin),
                     Q = rabbit_volatile_queue:new(QName),
                     Spec = #{no_ack => true,

deps/rabbit/src/rabbit_db_queue.erl

@@ -797,11 +797,16 @@ update_durable_in_khepri(UpdateFun, FilterFun) ->
 %%
 %% @private
 
-exists(QName) ->
-    rabbit_khepri:handle_fallback(
-      #{mnesia => fun() -> exists_in_mnesia(QName) end,
-        khepri => fun() -> exists_in_khepri(QName) end
-       }).
+exists(#resource{name = NameBin} = Name) ->
+    case rabbit_volatile_queue:is(NameBin) of
+        true ->
+            rabbit_volatile_queue:exists(Name);
+        false ->
+            rabbit_khepri:handle_fallback(
+              #{mnesia => fun() -> exists_in_mnesia(Name) end,
+                khepri => fun() -> exists_in_khepri(Name) end
+               })
+    end.
 
 exists_in_mnesia(QName) ->
     ets:member(?MNESIA_TABLE, QName).

deps/rabbit/src/rabbit_volatile_queue.erl

@@ -16,8 +16,8 @@
 -include_lib("rabbit_common/include/rabbit.hrl").
 
 -export([new/1,
+         new_name/0,
          is/1,
-         generate_name/1,
          key_from_name/1,
          pid_from_name/2,
          local_cast/2,
@@ -345,10 +345,10 @@ policy_apply_to_name() ->
 ff_enabled() ->
     rabbit_feature_flags:is_enabled('rabbitmq_4.2.0').
 
--spec generate_name(pid()) ->
+-spec new_name() ->
     rabbit_misc:resource_name().
-generate_name(Pid) ->
-    EncodedPid = encode_pid(Pid),
+new_name() ->
+    EncodedPid = encode_pid(self()),
     EncodedKey = base64:encode(rabbit_guid:gen()),
     <<?PREFIX, EncodedPid/binary, ".", EncodedKey/binary>>.