Use advertised TLS host setting in metadata frame

The rabbitmq_stream.advertised_tls_host setting is not used in the
metadata frame of the stream protocol, even if it is set. This commit
makes sure the setting is used if set.

References rabbitmq/rabbitmq-stream-java-client#803

Author: acogoluegnes

deps/rabbitmq_stream/src/rabbit_stream.erl

@@ -38,6 +38,7 @@
 
 -include("rabbit_stream_metrics.hrl").
 -include_lib("kernel/include/logger.hrl").
+-include_lib("rabbitmq_stream/src/rabbit_stream_utils.hrl").
 
 start(_Type, _Args) ->
     rabbit_stream_metrics:init(),
@@ -48,7 +49,7 @@ start(_Type, _Args) ->
     rabbit_stream_sup:start_link().
 
 tls_host() ->
-    case application:get_env(rabbitmq_stream, advertised_tls_host,
+    case application:get_env(rabbitmq_stream, ?K_AD_TLS_HOST,
                              undefined)
     of
         undefined ->
@@ -58,7 +59,7 @@ tls_host() ->
     end.
 
 host() ->
-    case application:get_env(rabbitmq_stream, advertised_host, undefined)
+    case application:get_env(rabbitmq_stream, ?K_AD_HOST, undefined)
     of
         undefined ->
             hostname_from_node();
@@ -79,7 +80,7 @@ hostname_from_node() ->
     end.
 
 port() ->
-    case application:get_env(rabbitmq_stream, advertised_port, undefined)
+    case application:get_env(rabbitmq_stream, ?K_AD_PORT, undefined)
     of
         undefined ->
             port_from_listener();
@@ -103,7 +104,7 @@ port_from_listener() ->
     end.
 
 tls_port() ->
-    case application:get_env(rabbitmq_stream, advertised_tls_port,
+    case application:get_env(rabbitmq_stream, ?K_AD_TLS_PORT,
                              undefined)
     of
         undefined ->

deps/rabbitmq_stream/src/rabbit_stream_reader.erl

@@ -1473,32 +1473,17 @@ handle_frame_pre_auth(Transport,
                                                      VirtualHost,
                                                      {socket, S},
                                                      #{}),
-            AdvertisedHost =
-                case TransportLayer of
-                    tcp ->
-                        rabbit_stream:host();
-                    ssl ->
-                        rabbit_stream:tls_host()
-                end,
-            AdvertisedPort =
-                case TransportLayer of
-                    tcp ->
-                        rabbit_data_coercion:to_binary(
-                            rabbit_stream:port());
-                    ssl ->
-                        rabbit_data_coercion:to_binary(
-                            rabbit_stream:tls_port())
-                end,
 
-            ConnectionProperties =
-                #{<<"advertised_host">> => AdvertisedHost,
-                  <<"advertised_port">> => AdvertisedPort},
+            AdHost = advertised_host(TransportLayer),
+            AdPort = rabbit_data_coercion:to_binary(advertised_port(TransportLayer)),
+            ConnProps = #{<<"advertised_host">> => AdHost,
+                          <<"advertised_port">> => AdPort},
 
             ?LOG_DEBUG("sending open response ok ~ts", [VirtualHost]),
             Frame =
                 rabbit_stream_core:frame({response, CorrelationId,
                                           {open, ?RESPONSE_CODE_OK,
-                                           ConnectionProperties}}),
+                                           ConnProps}}),
 
             send(Transport, S, Frame),
             %% FIXME check if vhost is alive (see rabbit_reader:is_vhost_alive/2)
@@ -2337,13 +2322,10 @@ handle_frame_post_auth(Transport,
                      Nodes0),
     NodeEndpoints =
         lists:foldr(fun(Node, Acc) ->
-                       PortFunction =
-                           case TransportLayer of
-                               tcp -> port;
-                               ssl -> tls_port
-                           end,
-                       Host = rpc:call(Node, rabbit_stream, host, []),
-                       Port = rpc:call(Node, rabbit_stream, PortFunction, []),
+                       HostFun = advertised_host_fun(TransportLayer),
+                       PortFun = advertised_port_fun(TransportLayer),
+                       Host = rpc:call(Node, rabbit_stream, HostFun, []),
+                       Port = rpc:call(Node, rabbit_stream, PortFun, []),
                        case {is_binary(Host), is_integer(Port)} of
                            {true, true} -> Acc#{Node => {Host, Port}};
                            _ ->
@@ -4077,3 +4059,21 @@ retry_sac_call(Call, N) ->
         R ->
             R
     end.
+
+advertised_host(Transport) ->
+    F = advertised_host_fun(Transport),
+    rabbit_stream:F().
+
+advertised_port(Transport) ->
+    F = advertised_port_fun(Transport),
+    rabbit_stream:F().
+
+advertised_host_fun(tcp) ->
+    host;
+advertised_host_fun(ssl) ->
+    tls_host.
+
+advertised_port_fun(tcp) ->
+    port;
+advertised_port_fun(ssl) ->
+    tls_port.

deps/rabbitmq_stream/src/rabbit_stream_utils.hrl

@@ -14,3 +14,7 @@
 %%
 
 -define(IS_INVALID_REF(Ref), is_binary(Ref) andalso byte_size(Ref) > 255).
+-define(K_AD_HOST, advertised_host).
+-define(K_AD_PORT, advertised_port).
+-define(K_AD_TLS_HOST, advertised_tls_host).
+-define(K_AD_TLS_PORT, advertised_tls_port).