Use ssl option when initialising data reader

Author: dcorbacho

deps/rabbitmq_stream/src/rabbit_stream_reader.erl

@@ -79,7 +79,8 @@
          stats_timer :: undefined | reference(),
          resource_alarm :: boolean(),
          send_file_oct ::
-             atomics:atomics_ref()}). % number of bytes sent with send_file (for metrics)
+             atomics:atomics_ref(),
+         transport :: 'tcp' | 'ssl'}). % number of bytes sent with send_file (for metrics)
 -record(configuration,
         {initial_credits :: integer(),
          credits_required_for_unblocking :: integer(),
@@ -157,7 +158,8 @@ init([KeepaliveSup,
       #{initial_credits := InitialCredits,
         credits_required_for_unblocking := CreditsRequiredBeforeUnblocking,
         frame_max := FrameMax,
-        heartbeat := Heartbeat}]) ->
+        heartbeat := Heartbeat,
+        transport := ConnTransport}]) ->
     process_flag(trap_exit, true),
     {ok, Sock} =
         rabbit_networking:handshake(Ref,
@@ -193,7 +195,8 @@ init([KeepaliveSup,
                                    connection_step = tcp_connected,
                                    frame_max = FrameMax,
                                    resource_alarm = false,
-                                   send_file_oct = SendFileOct},
+                                   send_file_oct = SendFileOct,
+                                   transport = ConnTransport},
             State =
                 #stream_connection_state{consumers = #{},
                                          blocked = false,
@@ -1364,7 +1367,8 @@ handle_frame_post_auth(Transport,
                                               StreamSubscriptions,
                                           virtual_host = VirtualHost,
                                           user = User,
-                                          send_file_oct = SendFileOct} =
+                                          send_file_oct = SendFileOct,
+                                          transport = ConnTransport} =
                            Connection,
                        #stream_connection_state{consumers = Consumers} = State,
                        {request, CorrelationId,
@@ -1417,7 +1421,7 @@ handle_frame_post_auth(Transport,
                                 {{?MODULE, QueueResource, SubscriptionId, self()}, []},
                             {ok, Segment} =
                                 osiris:init_reader(LocalMemberPid, OffsetSpec,
-                                                   CounterSpec),
+                                                   CounterSpec, ConnTransport),
                             rabbit_log:info("Next offset for subscription ~p is ~p",
                                             [SubscriptionId,
                                              osiris_log:next_offset(Segment)]),

deps/rabbitmq_stream/src/rabbit_stream_sup.erl

@@ -98,7 +98,7 @@ tcp_listener_spec([Address,
                                         SocketOpts,
                                         ranch_tcp,
                                         rabbit_stream_connection_sup,
-                                        Configuration,
+                                        Configuration#{transport => tcp},
                                         'stream',
                                         NumAcceptors,
                                         "Stream TCP listener").
@@ -112,7 +112,7 @@ ssl_listener_spec([Address,
                                         Address, SocketOpts ++ SslOpts,
                                         ranch_ssl,
                                         rabbit_stream_connection_sup,
-                                        Configuration,
+                                        Configuration#{transport => ssl},
                                         'stream/ssl',
                                         NumAcceptors,
                                         "Stream TLS listener").

deps/rabbitmq_stream/test/rabbit_stream_SUITE.erl

@@ -32,7 +32,7 @@ all() ->
 groups() ->
     [{single_node, [],
       [test_stream, test_stream_tls, test_gc_consumers, test_gc_publishers]},
-     {cluster, [], [test_stream, java]}].
+     {cluster, [], [test_stream, test_stream_tls, java]}].
 
 init_per_suite(Config) ->
     rabbit_ct_helpers:log_environment(),
@@ -61,7 +61,8 @@ init_per_group(cluster = Group, Config) ->
                                      [{rmq_nodes_count, 3},
                                       {rmq_nodename_suffix, Group},
                                       {tcp_ports_base}]),
-    rabbit_ct_helpers:run_setup_steps(Config2,
+    Config3 = rabbit_ct_helpers:set_config(Config2, {rabbitmq_ct_tls_verify, verify_none}),
+    rabbit_ct_helpers:run_setup_steps(Config3,
                                       [fun(StepConfig) ->
                                           rabbit_ct_helpers:merge_app_env(StepConfig,
                                                                           {aten,
@@ -342,6 +343,9 @@ test_deliver(Transport, S, SubscriptionId, Body, C0) ->
         Chunk,
     C.
 
+test_metadata_update_stream_deleted(S, Stream) ->
+    test_metadata_update_stream_deleted(gen_tcp, S, Stream).
+
 test_metadata_update_stream_deleted(Transport, S, Stream) ->
     StreamSize = byte_size(Stream),
     FrameSize = 2 + 2 + 2 + 2 + StreamSize,