Return consumer in send_chunks

Not a tuple. The passed-in consumer is updated with the returned
tuple, so we're better off returning the updated consumer record.

References #4368

(cherry picked from commit 90d110e)

Author: acogoluegnes

deps/rabbitmq_stream/src/rabbit_stream_reader.erl

@@ -42,7 +42,7 @@
          properties :: map()}).
 -record(consumer,
         {configuration :: #consumer_configuration{},
-         credit :: integer(),
+         credit :: non_neg_integer(),
          log :: osiris_log:state(),
          last_listener_offset = undefined :: undefined | osiris:offset()}).
 -record(stream_connection_state,
@@ -1018,18 +1018,7 @@ open(cast,
                                                                                   [Reason]),
                                                        %% likely a connection problem
                                                        Consumer;
-                                                   {{segment, Log1},
-                                                    {credit, Credit1},
-                                                    {last_listener_offset,
-                                                     LLO1}} ->
-                                                       Consumer#consumer{log =
-                                                                             Log1,
-                                                                         credit
-                                                                             =
-                                                                             Credit1,
-                                                                         last_listener_offset
-                                                                             =
-                                                                             LLO1}
+                                                   {ok, Csmr} -> Csmr
                                                end
                                        end,
                                    ConsumersAcc#{CorrelationId => Consumer1}
@@ -1857,13 +1846,9 @@ handle_frame_post_auth(Transport,
                                                                "peer",
                                                                []),
                                     throw({stop, normal});
-                                {{segment, Log1}, {credit, Credit1},
-                                 {last_listener_offset, LLO1}} ->
-                                    ConsumerState1 =
-                                        ConsumerState#consumer{log = Log1,
-                                                               credit = Credit1,
-                                                               last_listener_offset
-                                                                   = LLO1},
+                                {ok,
+                                 #consumer{log = Log1, credit = Credit1} =
+                                     ConsumerState1} ->
                                     Consumers1 =
                                         Consumers#{SubscriptionId =>
                                                        ConsumerState1},
@@ -1945,12 +1930,7 @@ handle_frame_post_auth(Transport,
                                                "peer",
                                                []),
                     throw({stop, normal});
-                {{segment, Log1}, {credit, Credit1},
-                 {last_listener_offset, LLO1}} ->
-                    Consumer1 =
-                        Consumer#consumer{log = Log1,
-                                          credit = Credit1,
-                                          last_listener_offset = LLO1},
+                {ok, Consumer1} ->
                     {Connection,
                      State#stream_connection_state{consumers =
                                                        Consumers#{SubscriptionId
@@ -2721,53 +2701,52 @@ send_file_callback(Transport,
 
 send_chunks(Transport,
             #consumer{credit = Credit, last_listener_offset = LastLstOffset} =
-                State,
+                Consumer,
             Counter) ->
-    send_chunks(Transport, State, Credit, LastLstOffset, Counter).
+    send_chunks(Transport, Consumer, Credit, LastLstOffset, Counter).
 
-send_chunks(_Transport,
-            #consumer{log = Log},
-            0,
-            LastLstOffset,
-            _Counter) ->
-    {{segment, Log}, {credit, 0}, {last_listener_offset, LastLstOffset}};
+send_chunks(_Transport, Consumer, 0, LastLstOffset, _Counter) ->
+    {ok,
+     Consumer#consumer{credit = 0, last_listener_offset = LastLstOffset}};
 send_chunks(Transport,
-            #consumer{log = Log} = State,
+            #consumer{log = Log} = Consumer,
             Credit,
             LastLstOffset,
             Counter) ->
     send_chunks(Transport,
-                State,
+                Consumer,
                 Log,
                 Credit,
                 LastLstOffset,
                 true,
                 Counter).
 
 send_chunks(_Transport,
-            _State,
-            Segment,
+            Consumer,
+            Log,
             0 = _Credit,
             LastLstOffset,
             _Retry,
             _Counter) ->
-    {{segment, Segment}, {credit, 0},
-     {last_listener_offset, LastLstOffset}};
+    {ok,
+     Consumer#consumer{log = Log,
+                       credit = 0,
+                       last_listener_offset = LastLstOffset}};
 send_chunks(Transport,
             #consumer{configuration = #consumer_configuration{socket = S}} =
-                State,
-            Segment,
+                Consumer,
+            Log,
             Credit,
             LastLstOffset,
             Retry,
             Counter) ->
-    case osiris_log:send_file(S, Segment,
-                              send_file_callback(Transport, State, Counter))
+    case osiris_log:send_file(S, Log,
+                              send_file_callback(Transport, Consumer, Counter))
     of
-        {ok, Segment1} ->
+        {ok, Log1} ->
             send_chunks(Transport,
-                        State,
-                        Segment1,
+                        Consumer,
+                        Log1,
                         Credit - 1,
                         LastLstOffset,
                         true,
@@ -2778,13 +2757,13 @@ send_chunks(Transport,
             {error, closed};
         {error, Reason} ->
             {error, Reason};
-        {end_of_stream, Segment1} ->
+        {end_of_stream, Log1} ->
             case Retry of
                 true ->
                     timer:sleep(1),
                     send_chunks(Transport,
-                                State,
-                                Segment1,
+                                Consumer,
+                                Log1,
                                 Credit,
                                 LastLstOffset,
                                 false,
@@ -2793,8 +2772,8 @@ send_chunks(Transport,
                     #consumer{configuration =
                                   #consumer_configuration{member_pid =
                                                               LocalMember}} =
-                        State,
-                    NextOffset = osiris_log:next_offset(Segment1),
+                        Consumer,
+                    NextOffset = osiris_log:next_offset(Log1),
                     LLO = case {LastLstOffset, NextOffset > LastLstOffset} of
                               {undefined, _} ->
                                   osiris:register_offset_listener(LocalMember,
@@ -2807,8 +2786,10 @@ send_chunks(Transport,
                               _ ->
                                   LastLstOffset
                           end,
-                    {{segment, Segment1}, {credit, Credit},
-                     {last_listener_offset, LLO}}
+                    {ok,
+                     Consumer#consumer{log = Log1,
+                                       credit = Credit,
+                                       last_listener_offset = LLO}}
             end
     end.