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

(cherry picked from commit 22a9593)

Conflicts:
	deps/rabbitmq_stream/src/rabbit_stream.erl
	deps/rabbitmq_stream/src/rabbit_stream_utils.hrl

Author: acogoluegnes

deps/rabbitmq_stream/src/rabbit_stream.erl

@@ -37,6 +37,7 @@
 -include_lib("rabbitmq_stream_common/include/rabbit_stream.hrl").
 
 -include("rabbit_stream_metrics.hrl").
+-include_lib("rabbitmq_stream/src/rabbit_stream_utils.hrl").
 
 start(_Type, _Args) ->
     rabbit_stream_metrics:init(),
@@ -47,7 +48,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 ->
@@ -57,7 +58,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();
@@ -78,7 +79,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();
@@ -102,7 +103,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

@@ -1470,32 +1470,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},
 
             rabbit_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)
@@ -2334,13 +2319,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}};
                            _ ->
@@ -4074,3 +4056,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

@@ -0,0 +1,20 @@
+%% The contents of this file are subject to the Mozilla Public License
+%% at https://www.mozilla.org/en-US/MPL/2.0/
+%%
+%% Software distributed under the License is distributed on an "AS IS"
+%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
+%% the License for the specific language governing rights and
+%% limitations under the License.
+%%
+%% The Original Code is RabbitMQ.
+%%
+%% The Initial Developer of the Original Code is Pivotal Software, Inc.
+%% Copyright (c) 2025 Broadcom. All Rights Reserved.
+%% The term “Broadcom” refers to Broadcom Inc. and/or its subsidiaries. All rights reserved.
+%%
+
+-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).