Simplify Direct Reply-To

This commit is no change in functionality and mostly deletes dead code.

1. Code targeting Erlang 22 and below is deleted since the mininmum
   required Erlang version is higher nowadays.
   "In OTP 23 distribution flag DFLAG_BIG_CREATION became mandatory. All
   pids are now encoded using NEW_PID_EXT, even external pids received
   as PID_EXT from older nodes."
   https://www.erlang.org/doc/apps/erts/erl_ext_dist.html#new_pid_ext
2. All v1 encoding and decoding of the Pid is deleted since the lower
   version RabbitMQ node supports the v2 encoding nowadays.

Author: ansd

deps/rabbit/src/pid_recomposition.erl

@@ -7,7 +7,6 @@
 
 -module(pid_recomposition).
 
-
 %% API
 -export([
     to_binary/1,
@@ -19,40 +18,23 @@
 -define(TTB_PREFIX, 131).
 
 -define(NEW_PID_EXT, 88).
--define(PID_EXT, 103).
 -define(ATOM_UTF8_EXT, 118).
 -define(SMALL_ATOM_UTF8_EXT, 119).
 
-%%
-%% API
-%%
-
 -spec decompose(pid()) -> #{atom() => any()}.
 decompose(Pid) ->
     from_binary(term_to_binary(Pid, [{minor_version, 2}])).
 
 -spec from_binary(binary()) -> #{atom() => any()}.
 from_binary(Bin) ->
-    PidData = case Bin of
-        %% Erlang 23+
-        <<?TTB_PREFIX, ?NEW_PID_EXT, Val0/binary>> -> Val0;
-        %% Erlang 22
-        <<?TTB_PREFIX, ?PID_EXT, Val1/binary>> -> Val1
-    end,
+    <<?TTB_PREFIX, ?NEW_PID_EXT, PidData/binary>> = Bin,
     {Node, Rest2} = case PidData of
         <<?ATOM_UTF8_EXT, AtomLen:16/integer, Node0:AtomLen/binary, Rest1/binary>> ->
             {Node0, Rest1};
         <<?SMALL_ATOM_UTF8_EXT, AtomLen/integer, Node0:AtomLen/binary, Rest1/binary>> ->
             {Node0, Rest1}
     end,
-    {ID, Serial, Creation} = case Rest2 of
-        %% NEW_PID_EXT on Erlang 23+
-        <<ID0:32/integer, Serial0:32/integer, Creation0:32/integer>> ->
-            {ID0, Serial0, Creation0};
-        %% PID_EXT on Erlang 22
-        <<ID1:32/integer, Serial1:32/integer, Creation1:8/integer>> ->
-            {ID1, Serial1, Creation1}
-    end,
+    <<ID:32/integer, Serial:32/integer, Creation:32/integer>> = Rest2,
     #{
         node     => binary_to_atom(Node, utf8),
         id       => ID,
@@ -62,9 +44,16 @@ from_binary(Bin) ->
 
 -spec to_binary(#{atom() => any()}) -> binary().
 to_binary(#{node := Node, id := ID, serial := Serial, creation := Creation}) ->
-    BinNode = atom_to_binary(Node, utf8),
+    BinNode = atom_to_binary(Node),
     NodeLen = byte_size(BinNode),
-    <<?TTB_PREFIX:8/unsigned, ?NEW_PID_EXT:8/unsigned, ?ATOM_UTF8_EXT:8/unsigned, NodeLen:16/unsigned, BinNode/binary, ID:32, Serial:32, Creation:32>>.
+    <<?TTB_PREFIX:8/unsigned,
+      ?NEW_PID_EXT:8/unsigned,
+      ?ATOM_UTF8_EXT:8/unsigned,
+      NodeLen:16/unsigned,
+      BinNode/binary,
+      ID:32,
+      Serial:32,
+      Creation:32>>.
 
 -spec recompose(#{atom() => any()}) -> pid().
 recompose(M) ->

deps/rabbit/src/rabbit_channel.erl

@@ -297,22 +297,11 @@ send_command(Pid, Msg) ->
 
 -spec deliver_reply(binary(), mc:state()) -> 'ok'.
 deliver_reply(<<"amq.rabbitmq.reply-to.", EncodedBin/binary>>, Message) ->
-    case rabbit_direct_reply_to:decode_reply_to_v2(EncodedBin,
-                                                   rabbit_nodes:all_running_with_hashes()) of
+    Nodes = rabbit_nodes:all_running_with_hashes(),
+    case rabbit_direct_reply_to:decode_reply_to(EncodedBin, Nodes) of
         {ok, Pid, Key} ->
-            delegate:invoke_no_result(Pid, {?MODULE, deliver_reply_local,
-                                            [Key, Message]});
-        {error, _} ->
-            deliver_reply_v1(EncodedBin, Message)
-    end.
-
--spec deliver_reply_v1(binary(), mc:state()) -> 'ok'.
-deliver_reply_v1(EncodedBin, Message) ->
-    %% the the original encoding function
-    case rabbit_direct_reply_to:decode_reply_to_v1(EncodedBin) of
-        {ok, V1Pid, V1Key} ->
-            delegate:invoke_no_result(V1Pid,
-                                      {?MODULE, deliver_reply_local, [V1Key, Message]});
+            delegate:invoke_no_result(
+              Pid, {?MODULE, deliver_reply_local, [Key, Message]});
         {error, _} ->
             ok
     end.
@@ -331,30 +320,19 @@ deliver_reply_local(Pid, Key, Message) ->
 declare_fast_reply_to(<<"amq.rabbitmq.reply-to">>) ->
     exists;
 declare_fast_reply_to(<<"amq.rabbitmq.reply-to.", EncodedBin/binary>>) ->
-    case rabbit_direct_reply_to:decode_reply_to_v2(EncodedBin, rabbit_nodes:all_running_with_hashes()) of
-        {error, _} ->
-            declare_fast_reply_to_v1(EncodedBin);
+    Nodes = rabbit_nodes:all_running_with_hashes(),
+    case rabbit_direct_reply_to:decode_reply_to(EncodedBin, Nodes) of
         {ok, Pid, Key} ->
             Msg = {declare_fast_reply_to, Key},
             rabbit_misc:with_exit_handler(
               rabbit_misc:const(not_found),
-              fun() -> gen_server2:call(Pid, Msg, infinity) end)
+              fun() -> gen_server2:call(Pid, Msg, infinity) end);
+        {error, _} ->
+            not_found
     end;
 declare_fast_reply_to(_) ->
     not_found.
 
-declare_fast_reply_to_v1(EncodedBin) ->
-    %% the the original encoding function
-    case rabbit_direct_reply_to:decode_reply_to_v1(EncodedBin) of
-        {ok, V1Pid, V1Key} ->
-            Msg = {declare_fast_reply_to, V1Key},
-            rabbit_misc:with_exit_handler(
-              rabbit_misc:const(not_found),
-              fun() -> gen_server2:call(V1Pid, Msg, infinity) end);
-        {error, _} ->
-            not_found
-    end.
-
 -spec list() -> [pid()].
 
 list() ->
@@ -1319,7 +1297,7 @@ handle_method(#'basic.consume'{queue        = <<"amq.rabbitmq.reply-to">>,
                                Other -> Other
                            end,
                     %% Precalculate both suffix and key
-                    {Key, Suffix} = rabbit_direct_reply_to:compute_key_and_suffix_v2(self()),
+                    {Key, Suffix} = rabbit_direct_reply_to:compute_key_and_suffix(self()),
                     Consumer = {CTag, Suffix, Key},
                     State1 = State#ch{reply_consumer = Consumer},
                     case NoWait of

deps/rabbit/src/rabbit_direct_reply_to.erl

@@ -7,45 +7,14 @@
 
 -module(rabbit_direct_reply_to).
 
-%% API
--export([
-    %% Original amq.rabbitmq.reply-to target channel encoding
-    compute_key_and_suffix_v1/1,
-    decode_reply_to_v1/1,
+-export([compute_key_and_suffix/1,
+         decode_reply_to/2]).
 
-    %% v2 amq.rabbitmq.reply-to target channel encoding
-    compute_key_and_suffix_v2/1,
-    decode_reply_to_v2/2
-]).
-
-%%
-%% API
-%%
-
--type decoded_pid_and_key() :: {ok, pid(), binary()}.
-
--spec compute_key_and_suffix_v1(pid()) -> {binary(), binary()}.
-%% This original pid encoding function produces values that exceed routing key length limit
-%% on nodes with long (say, 130+ characters) node names.
-compute_key_and_suffix_v1(Pid) ->
-    Key = base64:encode(rabbit_guid:gen()),
-    PidEnc = base64:encode(term_to_binary(Pid)),
-    Suffix = <<PidEnc/binary, ".", Key/binary>>,
-    {Key, Suffix}.
-
--spec decode_reply_to_v1(binary()) -> decoded_pid_and_key() | {error, any()}.
-decode_reply_to_v1(Bin) ->
-    case string:lexemes(Bin, ".") of
-        [PidEnc, Key] -> Pid = binary_to_term(base64:decode(PidEnc)),
-                         {ok, Pid, unicode:characters_to_binary(Key)};
-        _             -> {error, unrecognized_format}
-    end.
-
-
--spec compute_key_and_suffix_v2(pid()) -> {binary(), binary()}.
 %% This pid encoding function produces values that are of mostly fixed size
 %% regardless of the node name length.
-compute_key_and_suffix_v2(Pid) ->
+-spec compute_key_and_suffix(pid()) ->
+    {binary(), binary()}.
+compute_key_and_suffix(Pid) ->
     Key = base64:encode(rabbit_guid:gen()),
 
     PidParts0 = #{node := Node} = pid_recomposition:decompose(Pid),
@@ -61,19 +30,22 @@ compute_key_and_suffix_v2(Pid) ->
     Suffix = <<RecomposedEncoded/binary, ".", Key/binary>>,
     {Key, Suffix}.
 
--spec decode_reply_to_v2(binary(), #{non_neg_integer() => node()}) -> decoded_pid_and_key() | {error, any()}.
-decode_reply_to_v2(Bin, CandidateNodes) ->
+-spec decode_reply_to(binary(), #{non_neg_integer() => node()}) ->
+    {ok, pid(), binary()} | {error, any()}.
+decode_reply_to(Bin, CandidateNodes) ->
     try
         [PidEnc, Key] = binary:split(Bin, <<".">>),
         RawPidBin = base64:decode(PidEnc),
         PidParts0 = #{node := ShortenedNodename} = pid_recomposition:from_binary(RawPidBin),
         {_, NodeHash} = rabbit_nodes_common:parts(ShortenedNodename),
         case maps:get(list_to_integer(NodeHash), CandidateNodes, undefined) of
-            undefined -> {error, target_node_not_found};
+            undefined ->
+                {error, target_node_not_found};
             Candidate ->
                 PidParts = maps:update(node, Candidate, PidParts0),
                 {ok, pid_recomposition:recompose(PidParts), Key}
         end
     catch
-        error:_ -> {error, unrecognized_format}
+        error:_ ->
+            {error, unrecognized_format}
     end.

deps/rabbit/test/rabbit_direct_reply_to_prop_SUITE.erl

@@ -8,7 +8,7 @@
 
 all() ->
     [
-     decode_reply_to_v2
+     decode_reply_to
     ].
 
 init_per_suite(Config) ->
@@ -32,7 +32,7 @@ end_per_testcase(_TestCase, _Config) ->
 %%% Tests %%%
 
 
-decode_reply_to_v2(Config) ->
+decode_reply_to(Config) ->
     rabbit_ct_proper_helpers:run_proper(
       fun() -> prop_decode_reply_to(Config) end,
       [],
@@ -61,9 +61,9 @@ prop_decode_reply_to(_) ->
             NonB64 = <<0, Random/binary>>, 
 
             {ok, pid_recomposition:recompose(PidParts), Key} =:=
-                rabbit_direct_reply_to:decode_reply_to_v2(IxBin, NodeMap)
+                rabbit_direct_reply_to:decode_reply_to(IxBin, NodeMap)
             andalso {error, target_node_not_found} =:=
-                rabbit_direct_reply_to:decode_reply_to_v2(IxBin, NoNodeMap)
+                rabbit_direct_reply_to:decode_reply_to(IxBin, NoNodeMap)
             andalso {error, unrecognized_format} =:=
-                rabbit_direct_reply_to:decode_reply_to_v2(NonB64, NodeMap)
+                rabbit_direct_reply_to:decode_reply_to(NonB64, NodeMap)
         end).